DEV Community: cylon

[open source] based-firewalld Linux iptables UI&firewalld UI distributed management platform

cylon — Sat, 22 Apr 2023 17:41:33 +0000

Quick Deployment：Uranus

Best firewalld UI

If you think the project fits your usage scenario, please give me a 🌟🌟🌟!!!

Uranus Gateway

The mission of Uranus is to transform iptables into a tool with similar security group of aliyun.

Uranus Gateway It is a web UI and openAPI for distributed management of iptables rules on linux machines. is based-firewalld, which converts all dbus API translate to rest api , essentially for unified management of all machines in the The iptables rules of enterprise hosts, make the manipulation unified and standardized, no longer have to worry about operational errors, no longer have to worry about difficult management.

Project Show

Architecture

Instantly view firewall status and rollback restart operations

You can also add all firewalld rules on Uranus

It also supports the NAT function of iptables, which can be turned on and off with one click

And supports rich rules, such as traffic speed limit, network auditing, etc.

auto discovery

template management

Of course, the most important thing is not the function of the firewalld base itself

Special feature

Support for delayed tasks：For example, if the Internet is banned during the Golden Week holiday(China biggest holiday), then the effective period of the rule is 10-1 0:0:00 ~ 10-8 0:0:00, and there is no need to operate on October 1. You can set it to take effect on October 1 before leaving get off work can
Support automatic discovery：Similar to zabbix host automatic discovery function
No need to deploy additional agent side：Because firewalld is a function provided by common Linux distributions, such as centos 7 8 9, debian 10 11, ubuntu20 22 and other distributions can be installed with one click, then this system does not need to install any additional agent
zone is converted to a template： firewalld has the concept of zone, here I abstract the zone into a template, that is, by applying the template, a group of templates can be generated and applied in batches for a large number of hosts, so as to realize the dynamic switching of the firewall
SSO/UUAP： Support openldap as the backend, providing one-click authentication for enterprises
Timing takes effect： Both iptables and firewalld bases provide rules to take effect at regular intervals, so you can use this rule to complete scheduled tasks
docker/kubernetes deployment： Can be quickly deployed on any end for management
declarative task： Can support a large number of rules issued, provided that your host is online
Modular start： You can choose the function you want to start
asynchronous task： Declarative API that supports a large number of asynchronous operations

Let’s explore more functions quickly. In essence, Uranus does not recommend UI processing, because the functions are relatively simple, so using API can be better embedded in your management system

For more deployment problems, you can directly raise issue

future

Do you hope to manage in a distributed mode or a single node mode to support more security policies in the future? For example ban ip, anti-ddos and the like

【github开源】Linux iptables 界面化分布式管理平台 go vue，底座firewalld

cylon — Sat, 22 Apr 2023 17:28:53 +0000

快速部署：Uranus

目前最优的firewalld前端

如果你觉得项目符合你的使用场景, 劳烦大陆点个 🌟🌟🌟 吧!!! 好人一生平安!!!

Uranus Gateway

Uranus使命是将iptables转化为类安全组功能的一个工具

Uranus Gateway 是一个分布式管理linux主机上的iptables规则的web UI与open API，在linux上使用的底座为firewalld，将firewalld的所有的功能转化为 rest api风格，本质上是为了统一的管理企业内所有主机的iptables规则，使得操作统一化，规范化，再也不用担心操作失误了，再也不用担心管理难了

环境的展示

架构图

可以即时查看防火墙状态以及回滚重启操作

也可以添加firewalld上所有规则，

也支持iptables的nat功能，可以一键开启关闭

并支持富规则，例如流量限速，网络行文审计等

自动发现

模板管理

当然最重要的并不是本身firewalld底座的功能

特色功能

支持延迟任务：例如十一黄金周假期禁网，那么规则生效期为10-1 0:0:00 ~ 10-8 0:0:00，无需10月1日操作了，可以在下班前设置10月1日生效即可
支持自动发现：类似zabbix主机自动发现功能
无需额外部署agent端：因为firewalld是linux常见发行版提供的功能，例如centos 7 8 9，debian 10 11, ubuntu20 22等发行版都可以一键安装，那么这个系统就等于不需要在额外安装任何agent了
zone转化为模板：firewalld存在zone的概念，这里我将zone抽象化为模板，就是通过应用模板可以为大量主机批量生成并应用一组模板，实现防火墙的动态切换
SSO/UUAP：支持openldap作为后端，为企业提供一键认证
定时生效：iptables与firewalld底座均提供了规则定时生效，那么可以利用该规则完成定时的任务
docker/kubernetes一键部署：可以快速部署在任意端来进行管理
声明式任务：可支持规则的大量下发，前提是你的主机在线
模块化启动：可以选择自己想要的功能进行启动
异步任务：声明式API，支持大量异步操作

更多功能赶紧来探索吧，本质上Uranus不推荐进行UI处理，因为功能比较单一，所以使用API可以更好的嵌入你的管理系统中

更多部署上的问题，可以直接提issue即可

未来功能投票

你希望未来是以分布式模式进行管理还是单节点模式支持更多的安全策略呢？例如 ban ip，anti-ddos之类

kubernetes networking model

cylon — Wed, 17 Aug 2022 08:50:00 +0000

Overview

本文将探讨Kubernetes中的网络模型，以及对各种网络模型进行分析。

Underlay Network Model

什么是Underlay Network

底层网络 Underlay Network 顾名思义是指网络设备基础设施，如交换机，路由器, DWDM 使用网络介质将其链接成的物理网络拓扑，负责网络之间的数据包传输。

图：Underlay network topology
Source：https://community.cisco.com/t5/data-center-switches/understanding-underlay-and-overlay-networks/td-p/4295870

underlay network 可以是二层，也可以是三层；二层 underlay network 的典型例子是以太网 Ethernet，三层是 underlay network 的典型例子是互联网 Internet。

而工作与二层的技术是 vlan，工作在三层的技术是由 OSPF, BGP 等协议组成

kubernetes中的underlay network

在kubernetes中，underlay network 中比较典型的例子是通过将宿主机作为路由器设备而，Pod 的网络则通过学习成路由条目从而实现跨节点通讯。

图：underlay network topology in kubernetes

这种模型下典型的有 flannel 的 host-gw 模式与 calico BGP 模式。

flannel host-gw ^[1]

flannel host-gw 模式中每个Node需要在同一个二层网络中，并将Node作为一个路由器，跨节点通讯将通过路由表方式进行，这样方式下将网络模拟成一个underlay network。

图：layer2 ethernet topology
Source：https://www.auvik.com/franklyit/blog/layer-3-switches-layer-2/

Notes：因为是通过路由方式，集群的cidr至少要配置16，因为这样可以保证，跨节点的Node作为一层网络，同节点的Pod作为一个网络。如果不是这种用情况，路由表处于相同的网络中，会存在网络不可达

Calico BGP ^[2]

BGP（Border Gateway Protocol）是去中心化自治路由协议。它是通过维护IP路由表或'前缀'表来实现AS （Autonomous System）之间的可访问性，属于向量路由协议。

图：BGP network topology
Source：https://infocenter.nokia.com/public/7705SAR214R1A/index.jsp?topic=%2Fcom.sar.routing_protocols%

与 flannel 不同的是，Calico 提供了的 BGP 网络解决方案，在网络模型上，Calico 与 Flannel host-gw 是近似的，但在软件架构的实现上，flannel 使用 flanneld 进程来维护路由信息；而 Calico 是包含多个守护进程的，其中 Brid 进程是一个 BGP 的客户端与路由反射器(Router Reflector)，BGP 客户端负责从 Felix 中获取路由并分发到其他 BGP Peer，而反射器在BGP中起了优化的作用。在同一个IBGP中，BGP客户端仅需要和一个 RR 相连，这样减少了AS内部维护的大量的BGP连接。通常情况下，RR 是真实的路由设备，而 Bird 作为 BGP 客户端工作。

图：Calico Network Architecture
Source：https://www.cisco.com/c/en/us/td/docs/dcn/whitepapers/cisco-nx-os-calico-network-design.html

IPVLAN & MACVLAN ^[4]

IPVLAN 和 MACVLAN 是一种网卡虚拟化技术，两者之间的区别为， IPVLAN 允许一个物理网卡拥有多个IP地址，并且所有的虚拟接口用同一个MAC地址；而 MACVLAN 则是相反的，其允许同一个网卡拥有多个MAC地址，而虚拟出的网卡可以没有IP地址。

因为是网卡虚拟化技术，而不是网络虚拟化技术，本质上来说属于 Overlay network，这种方式在虚拟化环境中与Overlay network 相比最大的特点就是可以将Pod的网络拉平到Node网络同级，从而提供更高的性能、低延迟的网络接口。本质上来说其网络模型属于下图中第二个。

虚拟网桥：创建一个虚拟网卡对(veth pair)，一头栽容器内，一头栽宿主机的root namespaces内。这样一来容器内发出的数据包可以通过网桥直接进入宿主机网络栈，而发往容器的数据包也可以经过网桥进入容器。
多路复用：使用一个中间网络设备，暴露多个虚拟网卡接口，容器网卡都可以介入这个中间设备，并通过MAC/IP地址来区分packet应该发往哪个容器设备。
硬件交换，为每个Pod分配一个虚拟网卡，这样一来，Pod与Pod之间的连接关系就会变得非常清晰，因为近乎物理机之间的通信基础。如今大多数网卡都支持SR-IOV功能，该功能将单一的物理网卡虚拟成多个VF接口，每个VF接口都有单独的虚拟PCIe通道，这些虚拟的PCIe通道共用物理网卡的PCIe通道。

图：Virtual networking modes: bridging, multiplexing and SR-IOV
Source：https://thenewstack.io/hackers-guide-kubernetes-networking/

在kubernetes中 IPVLAN 这种网络模型下典型的CNI有，multus 与 danm。

multus

multus 是 intel 开源的CNI方案，是由传统的 cni 与 multus，并且提供了 SR-IOV CNI 插件使 K8s pod 能够连接到 SR-IOV VF 。这是使用了 IPVLAN/MACVLAN 的功能。

当创建新的Pod后，SR-IOV 插件开始工作。配置 VF 将被移动到新的 CNI 名称空间。该插件根据 CNI 配置文件中的 “name” 选项设置接口名称。最后将VF状态设置为UP。

下图是一个 Multus 和 SR-IOV CNI 插件的网络环境，具有三个接口的 pod。

eth0 是 flannel 网络插件，也是作为Pod的默认网络
VF 是主机的物理端口 ens2f0 的实例化。这是英特尔X710-DA4上的一个端口。在Pod端的 VF 接口名称为 south0 。
这个VF使用了 DPDK 驱动程序，此 VF 是从主机的物理端口 ens2f1 实例化出的。这个是英特尔® X710-DA4上另外一个端口。 Pod 内的 VF 接口名称为 north0。该接口绑定到 DPDK 驱动程序 vfio-pci 。

图：Mutus networking Architecture overlay and SR-IOV
Source：https://builders.intel.com/docs/networkbuilders/enabling_new_features_in_kubernetes_for_NFV.pdf

Notes：terminology

NIC：network interface card，网卡

SR-IOV：single root I/O virtualization，硬件实现的功能，允许各虚拟机间共享PCIe设备。

VF：Virtual Function，基于PF，与PF或者其他VF共享一个物理资源。

PF：PCIe Physical Function，拥有完全控制PCIe资源的能力

DPDK：Data Plane Development Kit

于此同时，也可以将主机接口直接移动到Pod的网络名称空间，当然这个接口是必须存在，并且不能是与默认网络使用同一个接口。这种情况下，在普通网卡的环境中，就直接将Pod网络与Node网络处于同一个平面内了。

图：Mutus networking Architecture overlay and ipvlan
Source：https://devopstales.github.io/kubernetes/multus/

danm

DANM是诺基亚开源的CNI项目，目的是将电信级网络引入kubernetes中，与multus相同的是，也提供了SR-IOV/DPDK 的硬件技术，并且支持IPVLAN.

Overlay Network Model

什么是Overlay

叠加网络是使用网络虚拟化技术，在 underlay 网络上构建出的虚拟逻辑网络，而无需对物理网络架构进行更改。本质上来说，overlay network 使用的是一种或多种隧道协议 (tunneling)，通过将数据包封装，实现一个网络到另一个网络中的传输，具体来说隧道协议关注的是数据包（帧）。

图：overlay network topology
Source：https://www.researchgate.net/figure/Example-Overlay-Network-built-on-top-of-an-Internet-style-Underlay_fig4_230774628

常见的网络隧道技术

通用路由封装 ( Generic Routing Encapsulation ) 用于将来自 IPv4/IPv6的数据包封装为另一个协议的数据包中，通常工作与L3网络层中。
VxLAN (Virtual Extensible LAN)，是一个简单的隧道协议，本质上是将L2的以太网帧封装为L4中UDP数据包的方法，使用 4789 作为默认端口。VxLAN 也是 VLAN 的扩展对于 4096（$2^{12}$ 位 VLAN ID）扩展为1600万（$2^{24}$ 位 VNID ）个逻辑网络。

这种工作在 overlay 模型下典型的有 flannel 与 calico 中的的 VxLAN, IPIP 模式。

IPIP

IP in IP 也是一种隧道协议，与 VxLAN 类似的是，IPIP 的实现也是通过Linux内核功能进行的封装。IPIP 需要内核模块 ipip.ko 使用命令查看内核是否加载IPIP模块lsmod | grep ipip ；使用命令modprobe ipip 加载。

图：A simple IPIP network workflow
Source：https://ssup2.github.io/theory_analysis/IPIP_GRE_Tunneling/

Kubernetes中 IPIP 与 VxLAN 类似，也是通过网络隧道技术实现的。与 VxLAN 差别就是，VxLAN 本质上是一个 UDP包，而 IPIP 则是将包封装在本身的报文包上。

图：IPIP in kubernetes

图：IPIP packet with wireshark unpack

Notes：公有云可能不允许IPIP流量，例如Azure

VxLAN

kubernetes中不管是 flannel 还是 calico VxLAN的实现都是使用Linux内核功能进行的封装，Linux 对 vxlan 协议的支持时间并不久，2012 年 Stephen Hemminger 才把相关的工作合并到 kernel 中，并最终出现在 kernel 3.7.0 版本。为了稳定性和很多的功能，你可以会看到某些软件推荐在 3.9.0 或者 3.10.0 以后版本的 kernel 上使用 VxLAN。

图：A simple VxLAN network topology

在kubernetes中vxlan网络，例如 flannel，守护进程会根据kubernetes的Node而维护 VxLAN，名称为 flannel.1 这是 VNID，并维护这个网络的路由，当发生跨节点的流量时，本地会维护对端 VxLAN 设备的MAC地址，通过这个地址可以知道发送的目的端，这样就可以封包发送到对端，收到包的对端 VxLAN设备 flannel.1 解包后得到真实的目的地址。

查看 Forwarding database 列表

$ bridge fdb 
26:5e:87:90:91:fc dev flannel.1 dst 10.0.0.3 self permanent

图：VxLAN in kubernetes

图：VxLAN packet with wireshark unpack

Notes：VxLAN使用的4789端口，wireshark应该是根据端口进行分析协议的，而flannel在linux中默认端口是8472，此时抓包仅能看到是一个UDP包。

通过上述的架构可以看出，隧道实际上是一个抽象的概念，并不是建立的真实的两端的隧道，而是通过将数据包封装成另一个数据包，通过物理设备传输后，经由相同的设备（网络隧道）进行解包实现网络的叠加。

weave vxlan ^[3]

weave也是使用了 VxLAN 技术完成的包的封装，这个技术在 weave 中称之为 fastdp (fast data path)，与 calico 和 flannel 中用到的技术不同的，这里使用的是 Linux 内核中的 openvswitch datapath module，并且weave对网络流量进行了加密。

图：weave fastdp network topology
Source：https://www.weave.works/docs/net/latest/concepts/fastdp-how-it-works/

Notes：fastdp工作在Linux 内核版本 3.12 及更高版本，如果低于此版本的例如CentOS7，weave将工作在用户空间，weave中称之为 sleeve mode

Reference

^[1] flannel host-gw

^[2] calico bgp networking

^[3] calico bgp networking

^[4] sriov network

^[5] danm

H3C HCL与WSL2共存

cylon — Thu, 11 Aug 2022 15:22:15 +0000

Background

由于Windows10 开启WSL2后无法和 eNSP 做到兼容，但是 H3C HCL 在版本 HCL_v2.1.2.1 提供了 VirtualBox 6.0.14 作为虚拟化后端，理论上来说可以做到 WSL2 与 HCL 共存。

并且开启了WSL2后并于其他虚拟化平台（VirtualBox, Vmvare）做到兼容的情况下，这种情况大部分禁止套娃（虚拟化下在虚拟化），通过安装虚拟机的方式再安装 eNSP 发现启动不报错，但是很长时间起不来。

Notes ^[1]：HCL官方给的建议是，对于windows7装的版本为HCL_v2.1.1；对于Windows10 并且开启了 Hype-v 或者 Dokcer-Desktop，推荐使用 HCL_v3.0.1.1

下载地址：HCL Download

安装过程

下载好安装时，直接下一步直至完成即可，VirtualBox已被内嵌至安装包内了。

图：HCL安装界面

Notes：如果需要抓包，自行安装Wireshark，安装好后，在HCL设置中配置 wireshark.exe 的路径即可

VirtualBox启用hyper-v支持 ^[2]

进入VirtualBox安装目录, 确定当前目录下存在VBoxManage.exe文件, 在当前目录打开powershell. 或者你将VBoxManage.exe所在目录加入环境变量, 任意路径下打开powershell.

# 或指定vbox所有虚拟系统开启
./VBoxManage.exe setextradata global "VBoxInternal/NEM/UseRing0Runloop" 0

开启后，HCL所有的设备就工作正常了，这种情况下也不用牺牲WSL2或者Dokcer-Desktop。因为eNSP官方没有再更新，导致hype-v与VirtualBox无法兼容，暂时无解。

Reference

^[1] H3C Cloud Lab

^[2] Windows 10 (2004) 启用wsl2, 并与VirtualBox 6.0+共存

巧用Prometheus来扩展kubernetes调度器

cylon — Mon, 08 Aug 2022 02:49:18 +0000

Overview

本文将深入讲解如何扩展 Kubernetes scheduler 中各个扩展点如何使用，与扩展scheduler的原理，这些是作为扩展 scheduler 的所需的知识点。最后会完成一个实验，记录网络流量的调度器。

kubernetes调度配置

kubernetes集群中允许运行多个不同的 scheduler ，也可以为Pod指定不同的调度器进行调度。在一般的Kubernetes调度教程中并没有提到这点，这也就是说，对于亲和性，污点等策略实际上并没有完全的使用kubernetes调度功能，在之前的文章中提到的一些调度插件，如基于端口占用的调度 NodePorts 等策略一般情况下是没有使用到的，本章节就是对这部分内容进行讲解，这也是作为扩展调度器的一个基础。

Scheduler Configuration ^[1]

kube-scheduler 提供了配置文件的资源，作为给 kube-scheduler 的配置文件，启动时通过 --config= 来指定文件。目前各个kubernetes版本中使用的 KubeSchedulerConfiguration 为，

1.21 之前版本使用 v1beta1
1.22 版本使用 v1beta2 ，但保留了 v1beta1
1.23, 1.24, 1.25 版本使用 v1beta3 ，但保留了 v1beta2，删除了 v1beta1

下面是一个简单的 kubeSchedulerConfiguration 示例，其中 kubeconfig 与启动参数 --kubeconfig 是相同的功效。而 kubeSchedulerConfiguration 与其他组件的配置文件类似，如 kubeletConfiguration 都是作为服务启动的配置文件。

apiVersion: kubescheduler.config.k8s.io/v1beta1
kind: KubeSchedulerConfiguration
clientConnection:
  kubeconfig: /etc/srv/kubernetes/kube-scheduler/kubeconfig

Notes: --kubeconfig 与 --config 是不可以同时指定的，指定了 --config 则其他参数自然失效 ^[2]

kubeSchedulerConfiguration使用

通过配置文件，用户可以自定义多个调度器，以及配置每个阶段的扩展点。而插件就是通过这些扩展点来提供在整个调度上下文中的调度行为。

下面配置是对于配置扩展点的部分的一个示例，关于扩展点的讲解可以参考kubernetes官方文档调度上下文部分

apiVersion: kubescheduler.config.k8s.io/v1beta1
kind: KubeSchedulerConfiguration
profiles:
  - plugins:
      score:
        disabled:
        - name: PodTopologySpread
        enabled:
        - name: MyCustomPluginA
          weight: 2
        - name: MyCustomPluginB
          weight: 1

Notes: 如果name="*" 的话，这种情况下将禁用/启用对应扩展点的所有插件

既然kubernetes提供了多调度器，那么对于配置文件来说自然支持多个配置文件，profile也是列表形式，只要指定多个配置列表即可，下面是多配置文件示例，其中，如果存在多个扩展点，也可以为每个调度器配置多个扩展点。

apiVersion: kubescheduler.config.k8s.io/v1beta2
kind: KubeSchedulerConfiguration
profiles:
  - schedulerName: default-scheduler
    plugins:
      preScore:
        disabled:
        - name: '*'
      score:
        disabled:
        - name: '*'
  - schedulerName: no-scoring-scheduler
    plugins:
      preScore:
        disabled:
        - name: '*'
      score:
        disabled:
        - name: '*'

scheduler调度插件 ^[3]

kube-scheduler 默认提供了很多插件作为调度方法，默认不配置的情况下会启用这些插件，如：

ImageLocality：调度将更偏向于Node存在容器镜像的节点。扩展点：score.
TaintToleration：实现污点与容忍度功能。扩展点：filter, preScore, score.
NodeName：实现调度策略中最简单的调度方法 NodeName 的实现。扩展点：filter.
NodePorts：调度将检查Node端口是否已占用。扩展点：preFilter, filter.
NodeAffinity：提供节点亲和性相关功能。扩展点：filter, score.
PodTopologySpread：实现Pod拓扑域的功能。扩展点：preFilter, filter, preScore, score.
NodeResourcesFit：该插件将检查节点是否拥有 Pod 请求的所有资源。使用以下三种策略之一：LeastAllocated （默认）MostAllocated 和 RequestedToCapacityRatio。扩展点：preFilter, filter, score.
VolumeBinding：检查节点是否有或是否可以绑定请求的卷. 扩展点：preFilter, filter, reserve, preBind, score.
VolumeRestrictions：检查安装在节点中的卷是否满足特定于卷提供程序的限制。扩展点：filter.
VolumeZone：检查请求的卷是否满足它们可能具有的任何区域要求。扩展点：filter.
InterPodAffinity：实现Pod 间的亲和性与反亲和性的功能。扩展点：preFilter, filter, preScore, score.
PrioritySort：提供基于默认优先级的排序。扩展点：queueSort.

对于更多配置文件使用案例可以参考官方给出的文档

如何扩展kube-scheduler ^[4]

当在第一次考虑编写调度程序时，通常会认为扩展 kube-scheduler 是一件非常困难的事情，其实这些事情 kubernetes 官方早就想到了，kubernetes为此在 1.15 版本引入了framework的概念，framework旨在使 scheduler 更具有扩展性。

framework 通过重新定义各扩展点，将其作为 plugins 来使用，并且支持用户注册 out of tree 的扩展，使其可以被注册到 kube-scheduler 中，下面将对这些步骤进行分析。

定义入口

scheduler 允许进行自定义，但是对于只需要引用对应的 NewSchedulerCommand，并且实现自己的 plugins 的逻辑即可。

import (
    scheduler "k8s.io/kubernetes/cmd/kube-scheduler/app"
)

func main() {
    command := scheduler.NewSchedulerCommand(
            scheduler.WithPlugin("example-plugin1", ExamplePlugin1),
            scheduler.WithPlugin("example-plugin2", ExamplePlugin2))
    if err := command.Execute(); err != nil {
        fmt.Fprintf(os.Stderr, "%v\n", err)
        os.Exit(1)
    }
}

而 NewSchedulerCommand 允许注入 out of tree plugins，也就是注入外部的自定义 plugins，这种情况下就无需通过修改源码方式去定义一个调度器，而仅仅通过自行实现即可完成一个自定义调度器。

// WithPlugin 用于注入out of tree plugins 因此scheduler代码中没有其引用。
func WithPlugin(name string, factory runtime.PluginFactory) Option {
    return func(registry runtime.Registry) error {
        return registry.Register(name, factory)
    }
}

插件实现

对于插件的实现仅仅需要实现对应的扩展点接口。下面通过内置插件进行分析

对于内置插件 NodeAffinity ,我们通过观察他的结构可以发现，实现插件就是实现对应的扩展点抽象 interface 即可。

定义插件结构体

其中 framework.FrameworkHandle 是提供了Kubernetes API与 scheduler 之间调用使用的，通过结构可以看出包含 lister，informer等等，这个参数也是必须要实现的。

type NodeAffinity struct {
    handle framework.FrameworkHandle
}

实现对应的扩展点

func (pl *NodeAffinity) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    if err != nil {
        return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
    }

    node := nodeInfo.Node()
    if node == nil {
        return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
    }

    affinity := pod.Spec.Affinity

    var count int64
    // A nil element of PreferredDuringSchedulingIgnoredDuringExecution matches no objects.
    // An element of PreferredDuringSchedulingIgnoredDuringExecution that refers to an
    // empty PreferredSchedulingTerm matches all objects.
    if affinity != nil && affinity.NodeAffinity != nil && affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution != nil {
        // Match PreferredDuringSchedulingIgnoredDuringExecution term by term.
        for i := range affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
            preferredSchedulingTerm := &affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution[i]
            if preferredSchedulingTerm.Weight == 0 {
                continue
            }

            // TODO: Avoid computing it for all nodes if this becomes a performance problem.
            nodeSelector, err := v1helper.NodeSelectorRequirementsAsSelector(preferredSchedulingTerm.Preference.MatchExpressions)
            if err != nil {
                return 0, framework.NewStatus(framework.Error, err.Error())
            }

            if nodeSelector.Matches(labels.Set(node.Labels)) {
                count += int64(preferredSchedulingTerm.Weight)
            }
        }
    }

    return count, nil
}

最后在通过实现一个 New 函数来提供注册这个扩展的方法。通过这个 New 函数可以在 main.go 中将其作为 out of tree plugins 注入到 scheduler 中即可

// New initializes a new plugin and returns it.
func New(_ runtime.Object, h framework.FrameworkHandle) (framework.Plugin, error) {
    return &NodeAffinity{handle: h}, nil
}

实验：基于网络流量的调度 ^[7]

通过上面阅读了解到了如何扩展 scheduler 插件，下面实验将完成一个基于流量的调度，通常情况下，网络一个Node在一段时间内使用的网络流量也是作为生产环境中很常见的情况。例如在配置均衡的多个主机中，主机A作为业务拉单脚本运行，主机B作为寻常服务运行。因为拉单需要下载大量数据，而硬件资源占用的却很少，此时，如果有Pod被调度到该节点上，那么可能双方业务都会收到影响（前端代理觉得这个节点连接数少会被大量调度，而拉单脚本因为网络带宽的占用降低了效能）。

实验环境

一个kubernetes集群，至少保证有两个节点。
提供的kubernetes集群都需要安装prometheus node_exporter，可以是集群内部的，也可以是集群外部的，这里使用的是集群外部的。
对 promQL 与 client_golang 有所了解

实验大致分为以下几个步骤：

定义插件API
- 插件命名为 NetworkTraffic
定义扩展点
- 这里使用了 Score 扩展点，并且定义评分的算法
定义分数获取途径（从prometheus指标中拿到对应的数据）
定义对自定义调度器的参数传入
将项目部署到集群中（集群内部署与集群外部署）
实验的结果验证

实验将仿照内置插件 nodeaffinity 完成代码编写，为什么选择这个插件，只是因为这个插件相对比较简单，并且与我们实验目的基本相同，其实其他插件也是同样的效果。

整个实验的代码上传至 github.com/CylonChau/customScheduler

实验开始

错误处理

在初始化项目时，go mod tidy 等操作时，会遇到大量下面的错误

go: github.com/GoogleCloudPlatform/spark-on-k8s-operator@v0.0.0-20210307184338-1947244ce5f4 requires
        k8s.io/apiextensions-apiserver@v0.0.0: reading k8s.io/apiextensions-apiserver/go.mod at revision v0.0.0: unknown revision v0.0.0

kubernetes issue #79384 ^[5] 中有提到这个问题，粗略浏览下没有说明为什么会出现这个问题，在最下方有个大佬提供了一个脚本，出现上述问题无法解决时直接运行该脚本后正常。

#!/bin/sh
set -euo pipefail

VERSION=${1#"v"}
if [ -z "$VERSION" ]; then
    echo "Must specify version!"
    exit 1
fi
MODS=($(
    curl -sS https://raw.githubusercontent.com/kubernetes/kubernetes/v${VERSION}/go.mod |
    sed -n 's|.*k8s.io/\(.*\) => ./staging/src/k8s.io/.*|k8s.io/\1|p'
))
for MOD in "${MODS[@]}"; do
    V=$(
        go mod download -json "${MOD}@kubernetes-${VERSION}" |
        sed -n 's|.*"Version": "\(.*\)".*|\1|p'
    )
    go mod edit "-replace=${MOD}=${MOD}@${V}"
done
go get "k8s.io/kubernetes@v${VERSION}"

定义插件API

通过上面内容描述了解到了定义插件只需要实现对应的扩展点抽象 interface ，那么可以初始化项目目录 pkg/networtraffic/networktraffice.go。

定义插件名称与变量

const Name = "NetworkTraffic"
var _ = framework.ScorePlugin(&NetworkTraffic{})

定义插件的结构体

type NetworkTraffic struct {
    // 这个作为后面获取node网络流量使用
    prometheus *PrometheusHandle
    // FrameworkHandle 提供插件可以使用的数据和一些工具。
    // 它在插件初始化时传递给 plugin 工厂类。
    // plugin 必须存储和使用这个handle来调用framework函数。
    handle framework.FrameworkHandle
}

定义扩展点

因为选用 Score 扩展点，需要定义对应的方法，来实现对应的抽象

func (n *NetworkTraffic) Score(ctx context.Context, state *framework.CycleState, p *corev1.Pod, nodeName string) (int64, *framework.Status) {
    // 通过promethes拿到一段时间的node的网络使用情况
    nodeBandwidth, err := n.prometheus.GetGauge(nodeName)
    if err != nil {
        return 0, framework.NewStatus(framework.Error, fmt.Sprintf("error getting node bandwidth measure: %s", err))
    }
    bandWidth := int64(nodeBandwidth.Value)
    klog.Infof("[NetworkTraffic] node '%s' bandwidth: %s", nodeName, bandWidth)
    return bandWidth, nil // 这里直接返回就行
}

接下来需要对结果归一化，这里就回到了调度框架中扩展点的执行问题上了，通过源码可以看出，Score 扩展点需要实现的并不只是这单一的方法。

// Run NormalizeScore method for each ScorePlugin in parallel.
parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
    pl := f.scorePlugins[index]
    nodeScoreList := pluginToNodeScores[pl.Name()]
    if pl.ScoreExtensions() == nil {
        return
    }
    status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
    if !status.IsSuccess() {
        err := fmt.Errorf("normalize score plugin %q failed with error %v", pl.Name(), status.Message())
        errCh.SendErrorWithCancel(err, cancel)
        return
    }
})

通过上面代码了解到，实现 Score 就必须实现 ScoreExtensions，如果没有实现则直接返回。而根据 nodeaffinity 中示例发现这个方法仅仅返回的是这个扩展点对象本身，而具体的归一化也就是真正进行打分的操作在 NormalizeScore 中。

// NormalizeScore invoked after scoring all nodes.
func (pl *NodeAffinity) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
    return pluginhelper.DefaultNormalizeScore(framework.MaxNodeScore, false, scores)
}

// ScoreExtensions of the Score plugin.
func (pl *NodeAffinity) ScoreExtensions() framework.ScoreExtensions {
    return pl
}

而在调度框架中，真正执行的操作的方法也是 NormalizeScore()

func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl framework.ScorePlugin, state *framework.CycleState, pod *v1.Pod, nodeScoreList framework.NodeScoreList) *framework.Status {
    if !state.ShouldRecordPluginMetrics() {
        return pl.ScoreExtensions().NormalizeScore(ctx, state, pod, nodeScoreList)
    }
    startTime := time.Now()
    status := pl.ScoreExtensions().NormalizeScore(ctx, state, pod, nodeScoreList)
    f.metricsRecorder.observePluginDurationAsync(scoreExtensionNormalize, pl.Name(), status, metrics.SinceInSeconds(startTime))
    return status
}

下面来实现对应的方法

在 NormalizeScore 中需要实现具体的选择node的算法，因为对node打分结果的区间为 $[0,100]$ ，所以这里实现的算法公式将为 $最高分 - (当前带宽 / 最高最高带宽 * 100)$，这样就保证了，带宽占用越大的机器，分数越低。

例如，最高带宽为200000，而当前Node带宽为140000，那么这个Node分数为：$max - \frac{140000}{200000}\times 100 = 100 - (0.7\times100)=30$

// 如果返回framework.ScoreExtensions 就需要实现framework.ScoreExtensions
func (n *NetworkTraffic) ScoreExtensions() framework.ScoreExtensions {
    return n
}

// NormalizeScore与ScoreExtensions是固定格式
func (n *NetworkTraffic) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *corev1.Pod, scores framework.NodeScoreList) *framework.Status {
    var higherScore int64
    for _, node := range scores {
        if higherScore < node.Score {
            higherScore = node.Score
        }
    }
    // 计算公式为，满分 - (当前带宽 / 最高最高带宽 * 100)
    // 公式的计算结果为，带宽占用越大的机器，分数越低
    for i, node := range scores {
        scores[i].Score = framework.MaxNodeScore - (node.Score * 100 / higherScore)
        klog.Infof("[NetworkTraffic] Nodes final score: %v", scores)
    }

    klog.Infof("[NetworkTraffic] Nodes final score: %v", scores)
    return nil
}

Notes：在kubernetes中最大的node数支持5000个，岂不是在获取最大分数时循环就占用了大量的性能，其实不必担心。scheduler 提供了一个参数 percentageOfNodesToScore。这个参数决定了这个部署循环的数量。更多的细节可以参考官方文档对这部分的说明 ^[6]

配置插件名称

为了使插件注册时候使用，还需要为其配置一个名称

// Name returns name of the plugin. It is used in logs, etc.
func (n *NetworkTraffic) Name() string {
    return Name
}

定义要传入的参数

网络插件的扩展中还存在一个 prometheusHandle，这个就是操作prometheus-server拿去指标的动作。

首先需要定义一个 PrometheusHandle 的结构体

type PrometheusHandle struct {
    deviceName string // 网络接口名称
    timeRange  time.Duration // 抓取的时间段
    ip         string // prometheus server的连接地址
    client     v1.API // 操作prometheus的客户端
}

有了结构就需要查询的动作和指标，对于指标来说，这里使用了 node_network_receive_bytes_total 作为获取Node的网络流量的计算方式。由于环境是部署在集群之外的，没有node的主机名，通过 promQL 获取，整个语句如下：

sum_over_time(node_network_receive_bytes_total{device="eth0"}[1s]) * on(instance) group_left(nodename) (node_uname_info{nodename="node01"})

整个 Prometheus 部分如下：

type PrometheusHandle struct {
    deviceName string
    timeRange  time.Duration
    ip         string
    client     v1.API
}

func NewProme(ip, deviceName string, timeRace time.Duration) *PrometheusHandle {
    client, err := api.NewClient(api.Config{Address: ip})
    if err != nil {
        klog.Fatalf("[NetworkTraffic] FatalError creating prometheus client: %s", err.Error())
    }
    return &PrometheusHandle{
        deviceName: deviceName,
        ip:         ip,
        timeRange:  timeRace,
        client:     v1.NewAPI(client),
    }
}

func (p *PrometheusHandle) GetGauge(node string) (*model.Sample, error) {
    value, err := p.query(fmt.Sprintf(nodeMeasureQueryTemplate, node, p.deviceName, p.timeRange))
    fmt.Println(fmt.Sprintf(nodeMeasureQueryTemplate, p.deviceName, p.timeRange, node))
    if err != nil {
        return nil, fmt.Errorf("[NetworkTraffic] Error querying prometheus: %w", err)
    }

    nodeMeasure := value.(model.Vector)
    if len(nodeMeasure) != 1 {
        return nil, fmt.Errorf("[NetworkTraffic] Invalid response, expected 1 value, got %d", len(nodeMeasure))
    }
    return nodeMeasure[0], nil
}

func (p *PrometheusHandle) query(promQL string) (model.Value, error) {
    // 通过promQL查询并返回结果
    results, warnings, err := p.client.Query(context.Background(), promQL, time.Now())
    if len(warnings) > 0 {
        klog.Warningf("[NetworkTraffic Plugin] Warnings: %v\n", warnings)
    }

    return results, err
}

配置调度器的参数

因为需要指定 prometheus 的地址，网卡名称，和获取数据的大小，故整个结构体如下，另外，参数结构必须遵循<Plugin Name>Args 格式的名称。

type NetworkTrafficArgs struct {
    IP         string `json:"ip"`
    DeviceName string `json:"deviceName"`
    TimeRange  int    `json:"timeRange"`
}

为了使这个类型的数据作为 KubeSchedulerConfiguration 可以解析的结构，还需要做一步操作，就是在扩展APIServer时扩展对应的资源类型。在这里kubernetes中提供两种方法来扩展 KubeSchedulerConfiguration 的资源类型。

一种是旧版中提供了 framework.DecodeInto 函数可以做这个操作

func New(plArgs *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
    args := Args{}
    if err := framework.DecodeInto(plArgs, &args); err != nil {
        return nil, err
    }
    ...
}

另外一种方式是必须实现对应的深拷贝方法，例如 NodeLabel 中的

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// NodeLabelArgs holds arguments used to configure the NodeLabel plugin.
type NodeLabelArgs struct {
    metav1.TypeMeta

    // PresentLabels should be present for the node to be considered a fit for hosting the pod
    PresentLabels []string
    // AbsentLabels should be absent for the node to be considered a fit for hosting the pod
    AbsentLabels []string
    // Nodes that have labels in the list will get a higher score.
    PresentLabelsPreference []string
    // Nodes that don't have labels in the list will get a higher score.
    AbsentLabelsPreference []string
}

最后将其注册到register中，整个行为与扩展APIServer是类似的

// addKnownTypes registers known types to the given scheme
func addKnownTypes(scheme *runtime.Scheme) error {
    scheme.AddKnownTypes(SchemeGroupVersion,
        &KubeSchedulerConfiguration{},
        &Policy{},
        &InterPodAffinityArgs{},
        &NodeLabelArgs{},
        &NodeResourcesFitArgs{},
        &PodTopologySpreadArgs{},
        &RequestedToCapacityRatioArgs{},
        &ServiceAffinityArgs{},
        &VolumeBindingArgs{},
        &NodeResourcesLeastAllocatedArgs{},
        &NodeResourcesMostAllocatedArgs{},
    )
    scheme.AddKnownTypes(schema.GroupVersion{Group: "", Version: runtime.APIVersionInternal}, &Policy{})
    return nil
}

Notes：对于生成深拷贝函数及其他文件，可以使用 kubernetes 代码库中的脚本 kubernetes/hack/update-codegen.sh

这里为了方便使用了 framework.DecodeInto 的方式。

项目部署

准备 scheduler 的 profile，可以看到，我们自定义的参数，就可以被识别为 KubeSchedulerConfiguration 的资源类型了。

apiVersion: kubescheduler.config.k8s.io/v1beta1
kind: KubeSchedulerConfiguration
clientConnection:
  kubeconfig: /mnt/d/src/go_work/customScheduler/scheduler.conf
profiles:
- schedulerName: custom-scheduler
  plugins:
    score:
      enabled:
      - name: "NetworkTraffic"
      disabled:
      - name: "*"
  pluginConfig:
    - name: "NetworkTraffic"
      args:
        ip: "http://10.0.0.4:9090"
        deviceName: "eth0"
        timeRange: 60

如果需要部署到集群内部，可以打包成镜像

FROM golang:alpine AS builder
MAINTAINER cylon
WORKDIR /scheduler
COPY ./ /scheduler
ENV GOPROXY https://goproxy.cn,direct
RUN \
    sed -i 's/dl-cdn.alpinelinux.org/mirrors.ustc.edu.cn/g' /etc/apk/repositories && \
    apk add upx  && \
    GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags "-s -w" -o scheduler main.go && \
    upx -1 scheduler && \
    chmod +x scheduler

FROM alpine AS runner
WORKDIR /go/scheduler
COPY --from=builder /scheduler/scheduler .
COPY --from=builder /scheduler/scheduler.yaml /etc/
VOLUME ["./scheduler"]

部署在集群内部所需的资源清单

apiVersion: v1
kind: ServiceAccount
metadata:
  name: scheduler-sa
  namespace: kube-system
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: scheduler
subjects:
  - kind: ServiceAccount
    name: scheduler-sa
    namespace: kube-system
roleRef:
  kind: ClusterRole
  name: system:kube-scheduler
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: custom-scheduler
  namespace: kube-system
  labels:
    component: custom-scheduler
spec:
  selector:
    matchLabels:
      component: custom-scheduler
  template:
    metadata:
      labels:
        component: custom-scheduler
    spec:
      serviceAccountName: scheduler-sa
      priorityClassName: system-cluster-critical
      containers:
        - name: scheduler
          image: cylonchau/custom-scheduler:v0.0.1
          imagePullPolicy: IfNotPresent
          command:
            - ./scheduler
            - --config=/etc/scheduler.yaml
            - --v=3
          livenessProbe:
            httpGet:
              path: /healthz
              port: 10251
            initialDelaySeconds: 15
          readinessProbe:
            httpGet:
              path: /healthz
              port: 10251

启动自定义 scheduler，这里通过简单的二进制方式启动，所以需要一个kubeconfig做认证文件

./main --logtostderr=true \
    --address=127.0.0.1 \
    --v=3 \
    --config=`pwd`/scheduler.yaml \
    --kubeconfig=`pwd`/scheduler.conf

启动后为了验证方便性，关闭了原来的 kube-scheduler 服务，因为原来的 kube-scheduler 已经作为HA中的master，所以不会使用自定义的 scheduler 导致pod pending。

验证结果

准备一个需要部署的Pod，指定使用的调度器名称

apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx-deployment
spec:
  selector:
    matchLabels:
      app: nginx
  replicas: 2 
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx:1.14.2
        ports:
        - containerPort: 80
      schedulerName: custom-scheduler

这里实验环境为2个节点的kubernetes集群，master与node01，因为master的服务比node01要多，这种情况下不管怎样，调度结果永远会被调度到node01上。

$ kubectl get pods -o wide
NAME                                READY   STATUS    RESTARTS   AGE   IP             NODE     NOMINATED NODE   READINESS GATES
nginx-deployment-69f76b454c-lpwbl   1/1     Running   0          43s   192.168.0.17   node01   <none>           <none>
nginx-deployment-69f76b454c-vsb7k   1/1     Running   0          43s   192.168.0.16   node01   <none>           <none>

而调度器的日志如下

I0808 01:56:31.098189   27131 networktraffic.go:83] [NetworkTraffic] node 'node01' bandwidth: %!s(int64=12541068340)
I0808 01:56:31.098461   27131 networktraffic.go:70] [NetworkTraffic] Nodes final score: [{master-machine 0} {node01 12541068340}]
I0808 01:56:31.098651   27131 networktraffic.go:70] [NetworkTraffic] Nodes final score: [{master-machine 0} {node01 71}]
I0808 01:56:31.098911   27131 networktraffic.go:73] [NetworkTraffic] Nodes final score: [{master-machine 0} {node01 71}]
I0808 01:56:31.099275   27131 default_binder.go:51] Attempting to bind default/nginx-deployment-69f76b454c-vsb7k to node01
I0808 01:56:31.101414   27131 eventhandlers.go:225] add event for scheduled pod default/nginx-deployment-69f76b454c-lpwbl
I0808 01:56:31.101414   27131 eventhandlers.go:205] delete event for unscheduled pod default/nginx-deployment-69f76b454c-lpwbl
I0808 01:56:31.103604   27131 scheduler.go:609] "Successfully bound pod to node" pod="default/nginx-deployment-69f76b454c-lpwbl" node="no
de01" evaluatedNodes=2 feasibleNodes=2
I0808 01:56:31.104540   27131 scheduler.go:609] "Successfully bound pod to node" pod="default/nginx-deployment-69f76b454c-vsb7k" node="no
de01" evaluatedNodes=2 feasibleNodes=2

Reference

^[1] scheduling config

^[2] kube-scheduler

^[3] scheduling-plugins

^[4] custom scheduler plugins

^[5] ssues #79384

^[6] scheduler perf tuning

^[7] creating a kube-scheduler plugin

最小化安装debian11

cylon — Tue, 02 Aug 2022 11:50:12 +0000

Preparation

debian11几乎可以使用任何旧的计算机硬件，因为最小安装的要求非常低。以下是最低要求和推荐要求：

最低要求	推荐要求
存储：10 Gigabytes 内存：512 Megabytes CPU: 1 GigaHertz	存储：10 Gigabytes 内存：2 Gigabytes CPU: 1 GigaHertz or more

如何选择下载安装包

官网提供了安装包的下载，其中CD是网络安装，DVD是离线安装

debian官方下载页面

Notes：CD安装包很小，下载下来是 debian-11.4.0-amd64-netinst.iso 如名所示，这是一个网络安装包，所以推荐下载DVD部分，可以达到离线安装的效果

安装步骤

在界面中选择“Install”，安装将开始。如果图形化安装可以选择“Graphical install”，这里选择“Install”。

欢迎页面

完成后，系统将提示选择安装时的“语言”。选择喜欢的语言，然后按“Enter”。这里选择英文

选择语言页面

这将是接下来安装步骤

安装步骤概述

选择位置与键盘布局

选择区域

下面部署时选择键盘布局：中国大陆使用的键盘布局是美国-英语，不要选择英国-英语之类，布局是不一样的，会存在按键输出的结果会不同

选择键盘布局

完成上述操作后，将开始加载镜像。等待扫描完成。。。。

等待扫描组件

设置主机名和域名

这步骤中将配置一个“主机名”。与一个“域”名称。

配置主机名

“域” 可以选择留空确定

配置域

完成上述操作后，安装程序将提示需要设置 root 密码。输入您的 root 密码，然后在重新输入以进行验证后继续。

设置Root密码

设置非ROOT用户名、账户和密码

下一步创建一个非ROOT用户，这个步骤是必须的，并为这个新创建的帐户分配一个密码。以下截图将描述将如何完成此操作。

配置普通用户

为这个用户配置密码

为普通用户配置密码

为普通用户配置密码——二次确认

设置时钟时区

Eastern 美东时间

Central 北美中部

Mountain 北美山区时区

Pacific 太平洋时区

Alaska 阿拉斯加夏令时间

Hawaii 夏威夷时区

Arizona 亞利桑时区

East Indiana 印第安纳时区

Samoa 萨摩亚时间

配置时区

对磁盘分区

此步骤磁盘进行分区。这里选择“手动”选项

选择分区模式

选择手动进行划分为所需的分区。

选择硬盘

创建新的分区表

创建分区表

选择空闲的空间进行分区

选择空闲空间

创建一个新分区

为/boot划分分区

最终划分的分区

这里选No就行，提示是指不使用swap分区，No就是继续，Yes将返回分区页面

对于swap分区的提示

创建新分区需要格式化，当前的分区将会被删除，如果是新磁盘选择Yes格式化分区

确认格式化，进行分区

Base System安装

这里等待安装基础系统

确认格式化，进行分区

几分钟后，安装后会弹出一个界面，这里会扫描其他的media，这里因为没有，选择No就行。

扫描其他媒介

离线安装

会扫描安装的媒介，这里也有提示，如果没有额外的媒介，可以跳过该步骤

扫描其他媒介

配置网络镜像，建议配置下，如果不需要No即可

配置网络镜像

接下来会弹出一个界面，请选择“Debian镜像国家”。这个是配置镜像地址的，选择自己的国家和镜像站即可

选择镜像国家

这里选择的是中国和中科大镜像

选择镜像地址

配置HTTP代理，不选择跳过

配置http代理

如果选择网络安装，到这步骤时安装程序现在将在选择相关的。首先，选择离您所在国家最近的位置。Debian 镜像位置和域后检索剩余的文件

这里提示有一个匿名调查，这里选No即可

匿名调查

根据要求调整安装

在检索过程中，系统将提示需要自行选择以下预定义软件中的一个或多个。最小化安装仅选择基础系统与SSH即可

安装组件选择

接下来等待安装即可

安装过程

Notes：选择了DVD ISO将离线完成安装，如果使用了CD ISO，将从互联网上检索包并安装，这个时间将很长。

其中会提示一个引导按章，直接Yes即可

到了这里即将安装完成

完成Debian11最小化安装

看到到这里已经完成了安装，按“Continue”继续重启后即可

完成安装

看到系统的引导界面

Enjoy 👏👏

深入解析kubernetes中的选举机制

cylon — Sun, 31 Jul 2022 14:41:04 +0000

Overview

在 Kubernetes的 kube-controller-manager , kube-scheduler, 以及使用 Operator 的底层实现 controller-rumtime 都支持高可用系统中的leader选举，本文将以理解 controller-rumtime （底层的实现是 client-go）中的leader选举以在kubernetes controller中是如何实现的。

Background

在运行 kube-controller-manager 时，是有一些参数提供给cm进行leader选举使用的，可以参考官方文档提供的参数来了解相关参数。

--leader-elect                               Default: true
--leader-elect-renew-deadline duration       Default: 10s
--leader-elect-resource-lock string          Default: "leases"
--leader-elect-resource-name string          Default: "kube-controller-manager"
--leader-elect-resource-namespace string     Default: "kube-system"
--leader-elect-retry-period duration         Default: 2s
...

本身以为这些组件的选举动作时通过etcd进行的，但是后面对 controller-runtime 学习时，发现并没有配置其相关的etcd相关参数，这就引起了对选举机制的好奇。怀着这种好奇心搜索了下有关于 kubernetes的选举，发现官网是这么介绍的，下面是对官方的说明进行一个通俗总结。simple leader election with kubernetes

通过阅读文章得知，kubernetes API 提供了一中选举机制，只要运行在集群内的容器，都是可以实现选举功能的。

Kubernetes API通过提供了两个属性来完成选举动作的

ResourceVersions：每个API对象唯一一个ResourceVersion

Annotations：每个API对象都可以对这些key进行注释

注：这种选举会增加APIServer的压力。也就对etcd会产生影响

那么有了这些信息之后，我们来看一下，在Kubernetes集群中，谁是cm的leader（我们提供的集群只有一个节点，所以本节点就是leader）

在Kubernetes中所有启用了leader选举的服务都会生成一个 EndPoint ，在这个 EndPoint 中会有上面提到的label（Annotations）来标识谁是leader。

$ kubectl get ep -n kube-system
NAME                      ENDPOINTS   AGE
kube-controller-manager   <none>      3d4h
kube-dns                              3d4h
kube-scheduler            <none>      3d4h

这里以 kube-controller-manager 为例，来看下这个 EndPoint 有什么信息

[root@master-machine ~]# kubectl describe ep kube-controller-manager -n kube-system
Name:         kube-controller-manager
Namespace:    kube-system
Labels:       <none>
Annotations:  control-plane.alpha.kubernetes.io/leader:
                {"holderIdentity":"master-machine_06730140-a503-487d-850b-1fe1619f1fe1","leaseDurationSeconds":15,"acquireTime":"2022-06-27T15:30:46Z","re...
Subsets:
Events:
  Type    Reason          Age    From                     Message
  ----    ------          ----   ----                     -------
  Normal  LeaderElection  2d22h  kube-controller-manager  master-machine_76aabcb5-49ff-45ff-bd18-4afa61fbc5af became leader
  Normal  LeaderElection  9m     kube-controller-manager  master-machine_06730140-a503-487d-850b-1fe1619f1fe1 became leader

可以看出 Annotations: control-plane.alpha.kubernetes.io/leader: 标出了哪个node是leader。

election in controller-runtime

controller-runtime 有关leader选举的部分在 pkg/leaderelection 下面，总共100行代码，我们来看下做了些什么？

可以看到，这里只提供了创建资源锁的一些选项

type Options struct {
    // 在manager启动时，决定是否进行选举
    LeaderElection bool
    // 使用那种资源锁 默认为租用 lease
    LeaderElectionResourceLock string
    // 选举发生的名称空间
    LeaderElectionNamespace string
    // 该属性将决定持有leader锁资源的名称
    LeaderElectionID string
}

通过 NewResourceLock 可以看到，这里是走的 client-go/tools/leaderelection下面，而这个leaderelection也有一个 example 来学习如何使用它。

通过 example 可以看到，进入选举的入口是一个 RunOrDie() 的函数

// 这里使用了一个lease锁，注释中说愿意为集群中存在lease的监听较少
lock := &resourcelock.LeaseLock{
    LeaseMeta: metav1.ObjectMeta{
        Name:      leaseLockName,
        Namespace: leaseLockNamespace,
    },
    Client: client.CoordinationV1(),
    LockConfig: resourcelock.ResourceLockConfig{
        Identity: id,
    },
}

// 开启选举循环
leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
    Lock: lock,
    // 这里必须保证拥有的租约在调用cancel()前终止，否则会仍有一个loop在运行
    ReleaseOnCancel: true,
    LeaseDuration:   60 * time.Second,
    RenewDeadline:   15 * time.Second,
    RetryPeriod:     5 * time.Second,
    Callbacks: leaderelection.LeaderCallbacks{
        OnStartedLeading: func(ctx context.Context) {
            // 这里填写你的代码，
            // usually put your code
            run(ctx)
        },
        OnStoppedLeading: func() {
            // 这里清理你的lease
            klog.Infof("leader lost: %s", id)
            os.Exit(0)
        },
        OnNewLeader: func(identity string) {
            // we're notified when new leader elected
            if identity == id {
                // I just got the lock
                return
            }
            klog.Infof("new leader elected: %s", identity)
        },
    },
})

到这里，我们了解了锁的概念和如何启动一个锁，下面看下，client-go都提供了那些锁。

在代码 tools/leaderelection/resourcelock/interface.go 定义了一个锁抽象，interface提供了一个通用接口，用于锁定leader选举中使用的资源。

type Interface interface {
    // Get 返回选举记录
    Get(ctx context.Context) (*LeaderElectionRecord, []byte, error)

    // Create 创建一个LeaderElectionRecord
    Create(ctx context.Context, ler LeaderElectionRecord) error

    // Update will update and existing LeaderElectionRecord
    Update(ctx context.Context, ler LeaderElectionRecord) error

    // RecordEvent is used to record events
    RecordEvent(string)

    // Identity 返回锁的标识
    Identity() string

    // Describe is used to convert details on current resource lock into a string
    Describe() string
}

那么实现这个抽象接口的就是，实现的资源锁，我们可以看到，client-go提供了四种资源锁

leaselock
configmaplock
multilock
endpointlock

leaselock

Lease是kubernetes控制平面中的通过ETCD来实现的一个Leases的资源，主要为了提供分布式租约的一种控制机制。相关对这个API的描述可以参考于：Lease 。

在Kubernetes集群中，我们可以使用如下命令来查看对应的lease

$ kubectl get leases -A
NAMESPACE         NAME                      HOLDER                                                AGE
kube-node-lease   master-machine            master-machine                                        3d19h
kube-system       kube-controller-manager   master-machine_06730140-a503-487d-850b-1fe1619f1fe1   3d19h
kube-system       kube-scheduler            master-machine_1724e2d9-c19c-48d7-ae47-ee4217b27073   3d19h

$ kubectl describe leases kube-controller-manager -n kube-system
Name:         kube-controller-manager
Namespace:    kube-system
Labels:       <none>
Annotations:  <none>
API Version:  coordination.k8s.io/v1
Kind:         Lease
Metadata:
  Creation Timestamp:  2022-06-24T11:01:51Z
  Managed Fields:
    API Version:  coordination.k8s.io/v1
    Fields Type:  FieldsV1
    fieldsV1:
      f:spec:
        f:acquireTime:
        f:holderIdentity:
        f:leaseDurationSeconds:
        f:leaseTransitions:
        f:renewTime:
    Manager:         kube-controller-manager
    Operation:       Update
    Time:            2022-06-24T11:01:51Z
  Resource Version:  56012
  Self Link:         /apis/coordination.k8s.io/v1/namespaces/kube-system/leases/kube-controller-manager
  UID:               851a32d2-25dc-49b6-a3f7-7a76f152f071
Spec:
  Acquire Time:            2022-06-27T15:30:46.000000Z
  Holder Identity:         master-machine_06730140-a503-487d-850b-1fe1619f1fe1
  Lease Duration Seconds:  15
  Lease Transitions:       2
  Renew Time:              2022-06-28T06:09:26.837773Z
Events:                    <none>

下面来看下leaselock的实现，leaselock会实现了作为资源锁的抽象

type LeaseLock struct {
    // LeaseMeta 就是类似于其他资源类型的属性，包含name ns 以及其他关于lease的属性
    LeaseMeta  metav1.ObjectMeta
    Client     coordinationv1client.LeasesGetter // Client 就是提供了informer中的功能
    // lockconfig包含上面通过 describe 看到的 Identity与recoder用于记录资源锁的更改
    LockConfig ResourceLockConfig
    // lease 就是 API中的Lease资源，可以参考下上面给出的这个API的使用
    lease      *coordinationv1.Lease
}

下面来看下leaselock实现了那些方法？

Get

Get 是从spec中返回选举的记录

func (ll *LeaseLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) {
    var err error
    ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Get(ctx, ll.LeaseMeta.Name, metav1.GetOptions{})
    if err != nil {
        return nil, nil, err
    }
    record := LeaseSpecToLeaderElectionRecord(&ll.lease.Spec)
    recordByte, err := json.Marshal(*record)
    if err != nil {
        return nil, nil, err
    }
    return record, recordByte, nil
}

// 可以看出是返回这个资源spec里面填充的值
func LeaseSpecToLeaderElectionRecord(spec *coordinationv1.LeaseSpec) *LeaderElectionRecord {
    var r LeaderElectionRecord
    if spec.HolderIdentity != nil {
        r.HolderIdentity = *spec.HolderIdentity
    }
    if spec.LeaseDurationSeconds != nil {
        r.LeaseDurationSeconds = int(*spec.LeaseDurationSeconds)
    }
    if spec.LeaseTransitions != nil {
        r.LeaderTransitions = int(*spec.LeaseTransitions)
    }
    if spec.AcquireTime != nil {
        r.AcquireTime = metav1.Time{spec.AcquireTime.Time}
    }
    if spec.RenewTime != nil {
        r.RenewTime = metav1.Time{spec.RenewTime.Time}
    }
    return &r
}

Create

Create 是在kubernetes集群中尝试去创建一个租约，可以看到，Client就是API提供的对应资源的REST客户端，结果会在Kubernetes集群中创建这个Lease

func (ll *LeaseLock) Create(ctx context.Context, ler LeaderElectionRecord) error {
    var err error
    ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Create(ctx, &coordinationv1.Lease{
        ObjectMeta: metav1.ObjectMeta{
            Name:      ll.LeaseMeta.Name,
            Namespace: ll.LeaseMeta.Namespace,
        },
        Spec: LeaderElectionRecordToLeaseSpec(&ler),
    }, metav1.CreateOptions{})
    return err
}

Update

Update 是更新Lease的spec

func (ll *LeaseLock) Update(ctx context.Context, ler LeaderElectionRecord) error {
    if ll.lease == nil {
        return errors.New("lease not initialized, call get or create first")
    }
    ll.lease.Spec = LeaderElectionRecordToLeaseSpec(&ler)

    lease, err := ll.Client.Leases(ll.LeaseMeta.Namespace).Update(ctx, ll.lease, metav1.UpdateOptions{})
    if err != nil {
        return err
    }

    ll.lease = lease
    return nil
}

RecordEvent

RecordEvent 是记录选举时出现的事件，这时候我们回到上部分在kubernetes集群中查看 ep 的信息时可以看到的event中存在 became leader 的事件，这里就是将产生的这个event添加到 meta-data 中。

func (ll *LeaseLock) RecordEvent(s string) {
   if ll.LockConfig.EventRecorder == nil {
      return
   }
   events := fmt.Sprintf("%v %v", ll.LockConfig.Identity, s)
   subject := &coordinationv1.Lease{ObjectMeta: ll.lease.ObjectMeta}
   // Populate the type meta, so we don't have to get it from the schema
   subject.Kind = "Lease"
   subject.APIVersion = coordinationv1.SchemeGroupVersion.String()
   ll.LockConfig.EventRecorder.Eventf(subject, corev1.EventTypeNormal, "LeaderElection", events)
}

到这里大致上了解了资源锁究竟是什么了，其他种类的资源锁也是相同的实现的方式，这里就不过多阐述了；下面的我们来看看选举的过程。

election workflow

选举的代码入口是在 leaderelection.go ，这里会继续上面的 example 向下分析整个选举的过程。

前面我们看到了进入选举的入口是一个 RunOrDie() 的函数，那么就继续从这里开始来了解。进入 RunOrDie，看到其实只有几行而已，大致上了解到了RunOrDie会使用提供的配置来启动选举的客户端，之后会阻塞，直到 ctx 退出，或停止持有leader的租约。

func RunOrDie(ctx context.Context, lec LeaderElectionConfig) {
    le, err := NewLeaderElector(lec)
    if err != nil {
        panic(err)
    }
    if lec.WatchDog != nil {
        lec.WatchDog.SetLeaderElection(le)
    }
    le.Run(ctx)
}

下面看下 NewLeaderElector 做了些什么？可以看到，LeaderElector是一个结构体，这里只是创建他，这个结构体提供了我们选举中所需要的一切（LeaderElector就是RunOrDie创建的选举客户端）。

func NewLeaderElector(lec LeaderElectionConfig) (*LeaderElector, error) {
    if lec.LeaseDuration <= lec.RenewDeadline {
        return nil, fmt.Errorf("leaseDuration must be greater than renewDeadline")
    }
    if lec.RenewDeadline <= time.Duration(JitterFactor*float64(lec.RetryPeriod)) {
        return nil, fmt.Errorf("renewDeadline must be greater than retryPeriod*JitterFactor")
    }
    if lec.LeaseDuration < 1 {
        return nil, fmt.Errorf("leaseDuration must be greater than zero")
    }
    if lec.RenewDeadline < 1 {
        return nil, fmt.Errorf("renewDeadline must be greater than zero")
    }
    if lec.RetryPeriod < 1 {
        return nil, fmt.Errorf("retryPeriod must be greater than zero")
    }
    if lec.Callbacks.OnStartedLeading == nil {
        return nil, fmt.Errorf("OnStartedLeading callback must not be nil")
    }
    if lec.Callbacks.OnStoppedLeading == nil {
        return nil, fmt.Errorf("OnStoppedLeading callback must not be nil")
    }

    if lec.Lock == nil {
        return nil, fmt.Errorf("Lock must not be nil.")
    }
    le := LeaderElector{
        config:  lec,
        clock:   clock.RealClock{},
        metrics: globalMetricsFactory.newLeaderMetrics(),
    }
    le.metrics.leaderOff(le.config.Name)
    return &le, nil
}

LeaderElector 是建立的选举客户端，

type LeaderElector struct {
    config LeaderElectionConfig // 这个的配置，包含一些时间参数，健康检查
    // recoder相关属性
    observedRecord    rl.LeaderElectionRecord
    observedRawRecord []byte
    observedTime      time.Time
    // used to implement OnNewLeader(), may lag slightly from the
    // value observedRecord.HolderIdentity if the transition has
    // not yet been reported.
    reportedLeader string
    // clock is wrapper around time to allow for less flaky testing
    clock clock.Clock
    // 锁定 observedRecord
    observedRecordLock sync.Mutex
    metrics leaderMetricsAdapter
}

可以看到 Run 实现的选举逻辑就是在初始化客户端时传入的三个 callback

func (le *LeaderElector) Run(ctx context.Context) {
    defer runtime.HandleCrash()
    defer func() { // 退出时执行callbacke的OnStoppedLeading
        le.config.Callbacks.OnStoppedLeading()
    }()

    if !le.acquire(ctx) {
        return
    }
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    go le.config.Callbacks.OnStartedLeading(ctx) // 选举时，执行 OnStartedLeading
    le.renew(ctx)
}

在 Run 中调用了 acquire，这个是通过一个loop去调用 tryAcquireOrRenew，直到ctx传递过来结束信号

func (le *LeaderElector) acquire(ctx context.Context) bool {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    succeeded := false
    desc := le.config.Lock.Describe()
    klog.Infof("attempting to acquire leader lease %v...", desc)
    // jitterUntil是执行定时的函数 func() 是定时任务的逻辑
    // RetryPeriod是周期间隔
    // JitterFactor 是重试系数，类似于延迟队列中的系数 （duration + maxFactor * duration）
    // sliding 逻辑是否计算在时间内
    // 上下文传递
    wait.JitterUntil(func() {
        succeeded = le.tryAcquireOrRenew(ctx)
        le.maybeReportTransition()
        if !succeeded {
            klog.V(4).Infof("failed to acquire lease %v", desc)
            return
        }
        le.config.Lock.RecordEvent("became leader")
        le.metrics.leaderOn(le.config.Name)
        klog.Infof("successfully acquired lease %v", desc)
        cancel()
    }, le.config.RetryPeriod, JitterFactor, true, ctx.Done())
    return succeeded
}

这里实际上选举动作在 tryAcquireOrRenew 中，下面来看下tryAcquireOrRenew；tryAcquireOrRenew 是尝试获得一个leader租约，如果已经获得到了，则更新租约；否则可以得到租约则为true，反之false

func (le *LeaderElector) tryAcquireOrRenew(ctx context.Context) bool {
    now := metav1.Now() // 时间
    leaderElectionRecord := rl.LeaderElectionRecord{ // 构建一个选举record
        HolderIdentity:       le.config.Lock.Identity(), // 选举人的身份特征，ep与主机名有关
        LeaseDurationSeconds: int(le.config.LeaseDuration / time.Second), // 默认15s
        RenewTime:            now, // 重新获取时间
        AcquireTime:          now, // 获得时间
    }

    // 1. 从API获取或创建一个recode，如果可以拿到则已经有租约，反之创建新租约
    oldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx)
    if err != nil {
        if !errors.IsNotFound(err) {
            klog.Errorf("error retrieving resource lock %v: %v", le.config.Lock.Describe(), err)
            return false
        }
        // 创建租约的动作就是新建一个对应的resource，这个lock就是leaderelection提供的四种锁，
        // 看你在runOrDie中初始化传入了什么锁
        if err = le.config.Lock.Create(ctx, leaderElectionRecord); err != nil {
            klog.Errorf("error initially creating leader election record: %v", err)
            return false
        }
        // 到了这里就已经拿到或者创建了租约，然后记录其一些属性，LeaderElectionRecord
        le.setObservedRecord(&leaderElectionRecord)

        return true
    }

    // 2. 获取记录检查身份和时间
    if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
        le.setObservedRecord(oldLeaderElectionRecord)

        le.observedRawRecord = oldLeaderElectionRawRecord
    }
    if len(oldLeaderElectionRecord.HolderIdentity) > 0 &&
        le.observedTime.Add(le.config.LeaseDuration).After(now.Time) &&
        !le.IsLeader() { // 不是leader，进行HolderIdentity比较，再加上时间，这个时候没有到竞选其，跳出
        klog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity)
        return false
    }

    // 3.我们将尝试更新。 在这里leaderElectionRecord设置为默认值。让我们在更新之前更正它。
    if le.IsLeader() { // 到这就说明是leader，修正他的时间
        leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
        leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
    } else { // LeaderTransitions 就是指leader调整（转变为其他）了几次，如果是，
        // 则为发生转变，保持原有值
        // 反之，则+1
        leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
    }
    // 完事之后更新APIServer中的锁资源，也就是更新对应的资源的属性信息
    if err = le.config.Lock.Update(ctx, leaderElectionRecord); err != nil {
        klog.Errorf("Failed to update lock: %v", err)
        return false
    }
    // setObservedRecord 是通过一个新的record来更新这个锁中的record
    // 操作是安全的，会上锁保证临界区仅可以被一个线程/进程操作
    le.setObservedRecord(&leaderElectionRecord)
    return true
}

summary

到这里，已经完整知道利用kubernetes进行选举的流程都是什么了；下面简单回顾下，上述leader选举所有的步骤：

首选创建的服务就是该服务的leader，锁可以为 lease , endpoint 等资源进行上锁
已经是leader的实例会不断续租，租约的默认值是15秒（leaseDuration）；leader在租约满时更新租约时间（renewTime）。
其他的follower，会不断检查对应资源锁的存在，如果已经有leader，那么则检查 renewTime，如果超过了租用时间（），则表明leader存在问题需要重新启动选举，直到有follower提升为leader。
而为了避免资源被抢占，Kubernetes API使用了 ResourceVersion 来避免被重复修改（如果版本号与请求版本号不一致，则表示已经被修改了，那么APIServer将返回错误）

Reference

Kubernetes 并发控制与数据一致性的实现原理

Controller manager 的高可用实现方式

deep dive into kubernetes simple leader election

让你的Nginx支持分布式追踪opentracing

cylon — Sun, 31 Jul 2022 14:40:00 +0000

Background

NGINX 是一个通用且流行的应用程序。也是最流行的 Web 服务器，它可用于提供静态文件内容，但也通常与其他服务一起用作分布式系统中的组件，在其中它用作反向代理、负载均衡或 API 网关。

分布式追踪 distributed tracing 是一种可用于分析与监控应用程序的机制，将追踪在从源到目的的整个过程中的单个请求，这与仅通过单个应用程序域来追踪请求的形式不同。

换句话说，我们可以说分布式追踪是对跨多个系统的多个请求的拼接。拼接通常由一个或多个相关 ID 完成，并且跟踪通常是一组记录的、跨所有系统的结构化日志事件，存储在一个中心位置。

在这种背景的情况下， OpenTracing 应运而生。OpenTracing 是一个与应用供应商无关的 API，它可帮助开发人员轻松地跟踪单一请求的域。目前有多种开源产品都支持 OpenTracing（例如，Jaeger, skywalking等），并将其作为一种检测分布式追踪的标准化方法。

本文将围绕，从0到1实现在nginx配置分布式追踪的架构的简单实例说明。本文实例使用的组件为

源码构建nginx-opentracing

准备nginx-opentracing

nginx-opentracing 仓库中可以看到，官方为每个nginx版本都提供了一个编译好的动态库（Nginx1.19.13+），我们可以直接拿来使用这个动态库，如果你想将这个利用Nginx 提供的编译参数 --add-module=/path/to/module 构建为nginx的内置功能的话，可能会出现一些问题，例如下面的一些错误：

ngx_http_opentracing_module.so/config was found

    /root/nginx-opentracing-0.25.0/opentracing//src/ngx_http_opentracing_module.cpp
In file included from /root/nginx-opentracing-0.25.0/opentracing//src/ngx_http_opentracing_module.cpp:1:0:
/root/nginx-opentracing-0.25.0/opentracing//src/load_tracer.h:3:38: fatal error: opentracing/dynamic_load.h: No such file or directory

根据 issue 中查询得知 nginx-opentracing 需要嵌入到nginx中，是需要一些 opentracing-cpp 因为对c++不熟，尝试调试很久还是上面的错误，故直接使用了官方提供的动态库。

准备jaeger-client-cpp

根据 nginx-opentracing 中提到的，还需要一个 jaeger-client-cpp 的 tracer 才可以正常运行（这也是作为jaeger架构中的角色）

来到 jaeger-client-cpp 看到Release提供的编译好的动态库已经很久了，而最新版都没有提供相应编译的版本，需要我们自己编译

说明：编译依赖CMake 3.3+，gcc 4.9.2+

我们的编译环境使用CentOS 7 默认gcc与CMake都符合要求需要自行编译两个的版本。

编译gcc

gcc下载地址：https://ftp.gnu.org/gnu/gcc/

cd gcc-5.4.0
./contrib/download_prerequisites

mkdir gcc-build-5.4.0
cd gcc-build-5.4.0

/usr/local/src/gcc-5.4.0/configure \
    --enable-checking=release \
    --enable-languages=c,c++ \
    --disable-multilib

make && make install

引用处理 ^{refer 1}

cd /usr/bin/
mv gcc gcc_back
mv g++ g++_back
ln -s /usr/local/bin/gcc gcc
ln -s /usr/local/bin/g++ g++

编译时遇到几个问题

/lib64/libstdc++.so.6: version GLIBCXX_3.4.20' not found

gcc 编译，libgcc动态库有改动，恢复原状即可

configure: error: C++ compiler missing or inoperational

make[2]: \*** [configure-stage1-libcpp] Error 1    
make[2]: Leaving directory `/home/clay/programming/C++/gcc-4.8.1'    
make[1]: \*** [stage1-bubble] Error 2    
make[1]: Leaving directory `/home/clay/programming/C++/gcc-4.8.1'    
make: \*** [all] Error 2

编译cmake

./configure --prefix=/path/to/app

make
make install

这里遇到一个小问题编译过程中遇到 [libstdc++.so.6: version GLIBCXX_3.4.20 not found

因为这里使用了自己编译的gcc版本，需要指定下动态库的路径 ^{refer 2}

LD_LIBRARY_PATH=/usr/local/lib64 ./configure --prefix=/usr/local/cmake

编译jaeger-client-cpp

这里根据官方提供的步骤操作即可

cd jaeger-client-cpp-0.9.0/
mkdir build
cd build
# 这里建议使用下特色上网，编译过程中会使用Hunter自动下载所需的依赖项
ALL_PROXY=http://x.0.0.x:10811 /usr/local/cmake/bin/cmake .. 
make

注：依赖项挺大的，下载时间可能很长，会hang主，只需等待结束即可

编译完成后 libjaegertracing.so.0.9.0 则是我们需要的

编译nginx

./configure \
    --user=web_www \
    --group=web_www \
    --with-pcre \
    --with-compat \
    --with-http_ssl_module  \
    --with-http_gzip_static_module \
    --prefix=/root/nginx  \
    --with-http_stub_status_module

--with-compat 必须加上，表面允许使用动态库，否则编译完在启动时会报下面的错误

nginx: [emerg] module "/root/nginx/conf/ngx_http_opentracing_module.so" is not binary compatible in /root/nginx/conf/nginx.conf:1

遇到的问题，cc nou found，这里只需将 gcc 软连接一份为 cc 即可

配置nginx

准备jaeger-client的配置

jaeger.json，参数的说明可以参考configuration

{
  "service_name": "nginx", // 服务名
  "sampler": {
    "type": "const",
    "param": 1
  },
  "reporter": {
    "localAgentHostPort": "jaeger:6831" // jaeger agent的地址
  },
  "headers": { // jaeger的默认的jaeger Baggage头设置
    "jaegerDebugHeader": "jaeger-debug-id",
    "jaegerBaggageHeader": "jaeger-baggage",
    "traceBaggageHeaderPrefix": "uberctx-"
  },
  "baggage_restrictions": {
    "denyBaggageOnInitializationFailure": false,
    "hostPort": ""
  }
}

在nginx中开启opentracing

对于更多的 nginx-opentracing 的配置说明参数可以参考 Reference.md

# 加载 OpenTracing 动态模块。
load_module conf/ngx_http_opentracing_module.so;
worker_processes  1;
user root root;

events {
    worker_connections  1024;
}
http {
    log_format opentracing '{"timestamp":"$time_iso8601",'
                       '"source":"$server_addr",'
                       '"hostname":"$hostname",'
                       '"ip":"$http_x_forwarded_for",'
                       '"traceID":"$opentracing_context_uber_trace_id",'
                       '"client":"$remote_addr",'
                       '"request_method":"$request_method",'
                       '"scheme":"$scheme",'
                       '"domain":"$server_name",'
                       '"referer":"$http_referer",'
                       '"request":"$request_uri",'
                       '"args":"$args",'
                       '"size":$body_bytes_sent,'
                       '"status": $status,'
                       '"responsetime":$request_time,'
                       '"upstreamtime":"$upstream_response_time",'
                       '"upstreamaddr":"$upstream_addr",'
                       '"http_user_agent":"$http_user_agent",'
                       '"https":"$https"'
                       '}';
    # 加载 tracer，这里使用的jaeger，需要传递配置文件
    opentracing_load_tracer conf/libjaegertracing.so conf/jaeger.json;
    # 启用 tracing 
    opentracing on; 
    # 设置tag，可选参数
    opentracing_tag http_user_agent $http_user_agent;
    include       mime.types;
    default_type  application/octet-stream;
    sendfile        on;
    keepalive_timeout  65;
    server {
        listen       80;
        server_name  localhost;
        location / {
            opentracing_operation_name $uri;
        opentracing_propagate_context;
            root   html;
            index  index.html index.htm;
        }
    access_log logs/access.log opentracing;
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }
    }
}

注：这里使用的 opentracing-nginx 的动态库为 ot16 ，linux-amd64-nginx-1.22.0-ot16-ngx_http_module.so.tgz ，另外一个版本不兼容，-t 检查语法时会提示

配置说明

对于每一个location都可以对其设置别名，这个就是 opentracing_operation_name 与 opentracing_location_operation_name 的区别

http {
...
    location = /upload/animal {
      opentracing_location_operation_name upload;
    ...

更多的配置说明可以参考 Tutorial.md

此时我们可以在jaeger上查看，可以看到 NGINX 的 span（因为这里只配置了NGINX，没有配置更多的后端）。

Reference

1 CentOS7 升级 GCC 到 5.4.0 版本

2 libstdc++.so.6: version GLIBCXX_3.4.20 not found

3 nginx load_module

深入解析Kubernetes admission webhooks

cylon — Sun, 31 Jul 2022 14:38:41 +0000

BACKGROUND

admission controllers的特点：

可定制性：准入功能可针对不同的场景进行调整。
可预防性：审计则是为了检测问题，而准入控制器可以预防问题发生
可扩展性：在kubernetes自有的验证机制外，增加了另外的防线，弥补了RBAC仅能对资源提供安全保证。

下图，显示了用户操作资源的流程，可以看出 admission controllers 作用是在通过身份验证资源持久化之前起到拦截作用。在准入控制器的加入会使kubernetes增加了更高级的安全功能。

图：Kubernetes API 请求的请求处理步骤图 Source：https://kubernetes.io/blog/2019/03/21/a-guide-to-kubernetes-admission-controllers/

这里找到一个大佬博客画的图，通过两张图可以很清晰的了解到admission webhook流程，与官方给出的不一样的地方在于，这里清楚地定位了kubernetes admission webhook 处于准入控制中，RBAC之后，push 之前。

图：Kubernetes API 请求的请求处理步骤图（详细） Source：https://www.armosec.io/blog/kubernetes-admission-controller/

两种控制器有什么区别？

根据官方提供的说法是

Mutating controllers may modify related objects to the requests they admit; validating controllers may not

从结构图中也可以看出，validating 是在持久化之前，而 Mutating 是在结构验证前，根据这些特性我们可以使用 Mutating 修改这个资源对象内容（如增加验证的信息），在 validating 中验证是否合法。

composition of admission controllers

kubernetes中的 admission controllers 由两部分组成：

内置在APIServer中的准入控制器 build-in list
特殊的控制器；也是内置在APIServer中，但提供一些自定义的功能
- MutatingAdmission
- ValidatingAdmission

Mutating 控制器可以修改他们处理的资源对象，Validating 控制器不会。当在任何一个阶段中的任何控制器拒绝这个了请求，则会立即拒绝整个请求，并将错误返回。

admission webhook

由于准入控制器是内置在 kube-apiserver 中的，这种情况下就限制了admission controller的可扩展性。在这种背景下，kubernetes提供了一种可扩展的准入控制器 extensible admission controllers，这种行为叫做动态准入控制 Dynamic Admission Control，而提供这个功能的就是 admission webhook 。

admission webhook 通俗来讲就是 HTTP 回调，通过定义一个http server，接收准入请求并处理。用户可以通过kubernetes提供的两种类型的 admission webhook，validating admission webhook 和 mutating admission webhook。来完成自定义的准入策略的处理。

webhook 就是

注：从上面的流程图也可以看出，admission webhook 也是有顺序的。首先调用mutating webhook，然后会调用validating webhook。

如何使用准入控制器

使用条件：kubernetes v1.16 使用 admissionregistration.k8s.io/v1 ；kubernetes v1.9 使用 admissionregistration.k8s.io/v1beta1。

如何在集群中开启准入控制器? ：查看kube-apiserver 的启动参数 --enable-admission-plugins ；通过该参数来配置要启动的准入控制器，如 --enable-admission-plugins=NodeRestriction 多个准入控制器以 , 分割，顺序无关紧要。反之使用 --disable-admission-plugins 参数可以关闭相应的准入控制器（Refer to apiserver opts）。

通过 kubectl 命令可以看到，当前kubernetes集群所支持准入控制器的版本



$ kubectl api-versions | grep admissionregistration.k8s.io/v1
admissionregistration.k8s.io/v1
admissionregistration.k8s.io/v1beta1

webhook工作原理

通过上面的学习，已经了解到了两种webhook的工作原理如下所示：

mutating webhook，会在持久化前拦截在 MutatingWebhookConfiguration 中定义的规则匹配的请求。MutatingAdmissionWebhook 通过向 mutating webhook 服务器发送准入请求来执行验证。

validaing webhook，会在持久化前拦截在 ValidatingWebhookConfiguration 中定义的规则匹配的请求。ValidatingAdmissionWebhook 通过将准入请求发送到 validating webhook server来执行验证。

那么接下来将从源码中看这个在这个工作流程中，究竟做了些什么？

资源类型

对于 1.9 版本之后，也就是 v1 版本，admission 被定义在 k8s.io\api\admissionregistration\v1\types.go ，大同小异，因为本地只有1.18集群，所以以这个讲解。

对于 Validating Webhook 来讲实现主要都在webhook中



type ValidatingWebhookConfiguration struct {
    // 每个api必须包含下列的metadata，这个是kubernetes规范，可以在注释中的url看到相关文档
    metav1.TypeMeta `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`
    // Webhooks在这里被表示为[]ValidatingWebhook，表示我们可以注册多个
    // +optional
    // +patchMergeKey=name
    // +patchStrategy=merge
    Webhooks []ValidatingWebhook `json:"webhooks,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,2,rep,name=Webhooks"`
}

webhook，则是对这种类型的webhook提供的操作、资源等。对于这部分不做过多的注释了，因为这里本身为kubernetes API资源，官网有很详细的例子与说明。这里更多字段的意思的可以参考官方 doc



type ValidatingWebhook struct {
    //  admission webhook的名词，Required
    Name string `json:"name" protobuf:"bytes,1,opt,name=name"`

    // ClientConfig 定义了与webhook通讯的方式 Required
    ClientConfig WebhookClientConfig `json:"clientConfig" protobuf:"bytes,2,opt,name=clientConfig"`

    // rule表示了webhook对于哪些资源及子资源的操作进行关注
    Rules []RuleWithOperations `json:"rules,omitempty" protobuf:"bytes,3,rep,name=rules"`

    // FailurePolicy 对于无法识别的value将如何处理，allowed/Ignore optional
    FailurePolicy *FailurePolicyType `json:"failurePolicy,omitempty" protobuf:"bytes,4,opt,name=failurePolicy,casttype=FailurePolicyType"`

    // matchPolicy 定义了如何使用“rules”列表来匹配传入的请求。
    MatchPolicy *MatchPolicyType `json:"matchPolicy,omitempty" protobuf:"bytes,9,opt,name=matchPolicy,casttype=MatchPolicyType"`
    NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty" protobuf:"bytes,5,opt,name=namespaceSelector"`
    SideEffects *SideEffectClass `json:"sideEffects" protobuf:"bytes,6,opt,name=sideEffects,casttype=SideEffectClass"`
    AdmissionReviewVersions []string `json:"admissionReviewVersions" protobuf:"bytes,8,rep,name=admissionReviewVersions"`
}

到这里了解了一个webhook资源的定义，那么这个如何使用呢？通过 Find Usages 找到一个 k8s.io/apiserver/pkg/admission/plugin/webhook/accessors.go 在使用它。这里没有注释，但在结构上可以看出，包含客户端与一系列选择器组成



type mutatingWebhookAccessor struct {
    *v1.MutatingWebhook
    uid               string
    configurationName string

    initObjectSelector sync.Once
    objectSelector     labels.Selector
    objectSelectorErr  error

    initNamespaceSelector sync.Once
    namespaceSelector     labels.Selector
    namespaceSelectorErr  error

    initClient sync.Once
    client     *rest.RESTClient
    clientErr  error
}

accessor 因为包含了整个webhookconfig定义的一些动作（这里个人这么觉得）。

accessor.go 下面有一个 GetRESTClient 方法，通过这里可以看出，这里做的就是使用根据 accessor 构造一个客户端。



func (m *mutatingWebhookAccessor) GetRESTClient(clientManager *webhookutil.ClientManager) (*rest.RESTClient, error) {
    m.initClient.Do(func() {
        m.client, m.clientErr = clientManager.HookClient(hookClientConfigForWebhook(m))
    })
    return m.client, m.clientErr
}

到这步骤已经没必要往下看了，因已经知道这里是请求webhook前的步骤了，下面就是何时请求了。

k8s.io\apiserver\pkg\admission\plugin\webhook\validating\dispatcher.go 下面有两个方法，Dispatch去请求我们自己定义的webhook



func (d *validatingDispatcher) Dispatch(ctx context.Context, attr admission.Attributes, o admission.ObjectInterfaces, hooks []webhook.WebhookAccessor) error {
    var relevantHooks []*generic.WebhookInvocation
    // Construct all the versions we need to call our webhooks
    versionedAttrs := map[schema.GroupVersionKind]*generic.VersionedAttributes{}
    for _, hook := range hooks {
        invocation, statusError := d.plugin.ShouldCallHook(hook, attr, o)
        if statusError != nil {
            return statusError
        }
        if invocation == nil {
            continue
        }
        relevantHooks = append(relevantHooks, invocation)
        // If we already have this version, continue
        if _, ok := versionedAttrs[invocation.Kind]; ok {
            continue
        }
        versionedAttr, err := generic.NewVersionedAttributes(attr, invocation.Kind, o)
        if err != nil {
            return apierrors.NewInternalError(err)
        }
        versionedAttrs[invocation.Kind] = versionedAttr
    }

    if len(relevantHooks) == 0 {
        // no matching hooks
        return nil
    }

    // Check if the request has already timed out before spawning remote calls
    select {
    case <-ctx.Done():
        // parent context is canceled or timed out, no point in continuing
        return apierrors.NewTimeoutError("request did not complete within requested timeout", 0)
    default:
    }

    wg := sync.WaitGroup{}
    errCh := make(chan error, len(relevantHooks))
    wg.Add(len(relevantHooks))
    // 循环所有相关的注册的hook
    for i := range relevantHooks {
        go func(invocation *generic.WebhookInvocation) {
            defer wg.Done()
            // invacation 中有一个 Accessor,Accessor注册了一个相关的webhookconfig
            // 也就是我们 kubectl -f 注册进来的那个webhook的相关配置
            hook, ok := invocation.Webhook.GetValidatingWebhook()
            if !ok {
                utilruntime.HandleError(fmt.Errorf("validating webhook dispatch requires v1.ValidatingWebhook, but got %T", hook))
                return
            }
            versionedAttr := versionedAttrs[invocation.Kind]
            t := time.Now()
            // 调用了callHook去请求我们自定义的webhook
            err := d.callHook(ctx, hook, invocation, versionedAttr)
            ignoreClientCallFailures := hook.FailurePolicy != nil && *hook.FailurePolicy == v1.Ignore
            rejected := false
            if err != nil {
                switch err := err.(type) {
                case *webhookutil.ErrCallingWebhook:
                    if !ignoreClientCallFailures {
                        rejected = true
                        admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionCallingWebhookError, 0)
                    }
                case *webhookutil.ErrWebhookRejection:
                    rejected = true
                    admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionNoError, int(err.Status.ErrStatus.Code))
                default:
                    rejected = true
                    admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionAPIServerInternalError, 0)
                }
            }
            admissionmetrics.Metrics.ObserveWebhook(time.Since(t), rejected, versionedAttr.Attributes, "validating", hook.Name)
            if err == nil {
                return
            }

            if callErr, ok := err.(*webhookutil.ErrCallingWebhook); ok {
                if ignoreClientCallFailures {
                    klog.Warningf("Failed calling webhook, failing open %v: %v", hook.Name, callErr)
                    utilruntime.HandleError(callErr)
                    return
                }

                klog.Warningf("Failed calling webhook, failing closed %v: %v", hook.Name, err)
                errCh <- apierrors.NewInternalError(err)
                return
            }

            if rejectionErr, ok := err.(*webhookutil.ErrWebhookRejection); ok {
                err = rejectionErr.Status
            }
            klog.Warningf("rejected by webhook %q: %#v", hook.Name, err)
            errCh <- err
        }(relevantHooks[i])
    }
    wg.Wait()
    close(errCh)

    var errs []error
    for e := range errCh {
        errs = append(errs, e)
    }
    if len(errs) == 0 {
        return nil
    }
    if len(errs) > 1 {
        for i := 1; i < len(errs); i++ {
            // TODO: merge status errors; until then, just return the first one.
            utilruntime.HandleError(errs[i])
        }
    }
    return errs[0]
}

callHook 可以理解为真正去请求我们自定义的webhook服务的动作



func (d *validatingDispatcher) callHook(ctx context.Context, h *v1.ValidatingWebhook, invocation *generic.WebhookInvocation, attr *generic.VersionedAttributes) error {
   if attr.Attributes.IsDryRun() {
      if h.SideEffects == nil {
         return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: fmt.Errorf("Webhook SideEffects is nil")}
      }
      if !(*h.SideEffects == v1.SideEffectClassNone || *h.SideEffects == v1.SideEffectClassNoneOnDryRun) {
         return webhookerrors.NewDryRunUnsupportedErr(h.Name)
      }
   }

   uid, request, response, err := webhookrequest.CreateAdmissionObjects(attr, invocation)
   if err != nil {
      return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
   }
   // 发生请求，可以看到，这里从上面的讲到的地方获取了一个客户端
   client, err := invocation.Webhook.GetRESTClient(d.cm)
   if err != nil {
      return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
   }
   trace := utiltrace.New("Call validating webhook",
      utiltrace.Field{"configuration", invocation.Webhook.GetConfigurationName()},
      utiltrace.Field{"webhook", h.Name},
      utiltrace.Field{"resource", attr.GetResource()},
      utiltrace.Field{"subresource", attr.GetSubresource()},
      utiltrace.Field{"operation", attr.GetOperation()},
      utiltrace.Field{"UID", uid})
   defer trace.LogIfLong(500 * time.Millisecond)

   // 这里设置超时，超时时长就是在yaml资源清单中设置的那个值
   if h.TimeoutSeconds != nil {
      var cancel context.CancelFunc
      ctx, cancel = context.WithTimeout(ctx, time.Duration(*h.TimeoutSeconds)*time.Second)
      defer cancel()
   }
   // 直接用post请求我们自己定义的webhook接口
   r := client.Post().Body(request)

   // if the context has a deadline, set it as a parameter to inform the backend
   if deadline, hasDeadline := ctx.Deadline(); hasDeadline {
      // compute the timeout
      if timeout := time.Until(deadline); timeout > 0 {
         // if it's not an even number of seconds, round up to the nearest second
         if truncated := timeout.Truncate(time.Second); truncated != timeout {
            timeout = truncated + time.Second
         }
         // set the timeout
         r.Timeout(timeout)
      }
   }

   if err := r.Do(ctx).Into(response); err != nil {
      return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
   }
   trace.Step("Request completed")

   result, err := webhookrequest.VerifyAdmissionResponse(uid, false, response)
   if err != nil {
      return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
   }

   for k, v := range result.AuditAnnotations {
      key := h.Name + "/" + k
      if err := attr.Attributes.AddAnnotation(key, v); err != nil {
         klog.Warningf("Failed to set admission audit annotation %s to %s for validating webhook %s: %v", key, v, h.Name, err)
      }
   }
   if result.Allowed {
      return nil
   }
   return &webhookutil.ErrWebhookRejection{Status: webhookerrors.ToStatusErr(h.Name, result.Result)}
}

走到这里基本上对 admission webhook 有了大致的了解，可以知道这个操作是由 apiserver 完成的。下面就实际操作下自定义一个webhook。

这里还有两个概念，就是请求参数 AdmissionRequest 和相应参数 AdmissionResponse，这些可以在 callHook 中看到，这两个参数被定义在 k8s.io\api\admission\v1\types.go ；这两个参数也就是我们在自定义 webhook 时需要处理接收到的body的结构，以及我们响应内容数据结构。

如何编写一个自定义的admission webhook

通过上面的学习了解到了，自定义的webhook就是做为kubernetes提供给用户两种admission controller来验证自定义业务的一个中间件 admission webhook。本质上他是一个HTTP Server，用户可以使用任何语言来完成这部分功能。当然，如果涉及到需要对kubernetes集群资源操作的话，还是建议使用kubernetes官方提供了SDK的编程语言来完成自定义的webhook。

那么完成一个自定义admission webhook需要两个步骤：

将相关的webhook config注册给kubernetes，也就是让kubernetes知道你的webhook
准备一个http server来处理 apiserver发过来验证的信息

注：这里使用go net/http包，本身不区分方法处理HTTP的何种请求，如果用其他框架实现的，如django，需要指定对应方法需要为POST

向kubernetes注册webhook对象

kubernetes提供的两种类型可自定义的准入控制器，和其他资源一样，可以利用资源清单，动态配置那些资源要被adminssion webhook处理。 kubernetes将这种形式抽象为两种资源：

ValidatingWebhookConfiguration
MutatingWebhookConfiguration

ValidatingAdmission



apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: "pod-policy.example.com"
webhooks:
- name: "pod-policy.example.com"
  rules:
  - apiGroups:   [""] # 拦截资源的Group "" 表示 core。"*" 表示所有。
    apiVersions: ["v1"] # 拦截资源的版本
    operations:  ["CREATE"] # 什么请求下拦截
    resources:   ["pods"]  # 拦截什么资源
    scope:       "Namespaced" # 生效的范围，cluster还是namespace "*"表示没有范围限制。
  clientConfig: # 我们部署的webhook服务，
    service: # service是在cluster-in模式下
      namespace: "example-namespace"
      name: "example-service"
      port: 443 # 服务的端口
      path: "/validate" # path是对应用于验证的接口
    # caBundle是提供给 admission webhook CA证书  
    caBundle: "Ci0tLS0tQk...<base64-encoded PEM bundle containing the CA that signed the webhook's serving certificate>...tLS0K"
  admissionReviewVersions: ["v1", "v1beta1"]
  sideEffects: None
  timeoutSeconds: 5 # 1-30s直接，表示请求api的超时时间

MutatingAdmission



apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: "valipod-policy.example.com"
webhooks:
- name: "valipod-policy.example.com"
  rules:
    - apiGroups:   ["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
      apiVersions: ["v1"] # 拦截资源的版本
      operations:  ["CREATE"] # 什么请求下拦截
      resources:   ["deployments"]  # 拦截什么资源
      scope:       "Namespaced" # 生效的范围，cluster还是namespace "*"表示没有范围限制。
  clientConfig: # 我们部署的webhook服务，
    url: "https://10.0.0.1:81/validate" # 这里是外部模式
    #      service: # service是在cluster-in模式下
    #        namespace: "default"
    #        name: "admission-webhook"
    #        port: 81 # 服务的端口
    #        path: "/mutate" # path是对应用于验证的接口
    # caBundle是提供给 admission webhook CA证书
    caBundle: "Ci0tLS0tQk...<base64-encoded PEM bundle containing the CA that signed the webhook's serving certificate>...tLS0K"
  admissionReviewVersions: ["v1"]
  sideEffects: None
  timeoutSeconds: 5 # 1-30s直接，表示请求api的超时时间

注：对于webhook，也可以引入外部的服务，并非必须部署到集群内部

对于外部服务来讲，需要 clientConfig 中的 service , 更换为 url ; 通过 url 参数可以将一个外部的服务引入



apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
...
webhooks:
- name: my-webhook.example.com
  clientConfig:
    url: "https://my-webhook.example.com:9443/my-webhook-path"
  ...

注：这里的url规则必须准守下列形式：

scheme://host:port/path

使用了url 时，这里不应填写集群内的服务

scheme 必须是 https，不能为http，这就意味着，引入外部时也需要

配置时使用了，?xx=xx 的参数也是不被允许的（官方说法是这样的，通过源码学习了解到因为会发送特定的请求体，所以无需管参数）

更多的配置可以参考kubernetes官方提供的 doc

准备一个webhook

让我们编写我们的 webhook server。将创建两个钩子，/mutate 与 /validate；

/mutate 将在创建deployment资源时，基于版本，给资源加上注释 webhook.example.com/allow: true
/validate 将对 /mutate 增加的 allow:true 那么则继续，否则拒绝。

这里为了方便，全部写在一起了，实际上不符合软件的设计。在kubernetes代码库中也提供了一个webhook server，可以参考他这个webhook server来学习具体要做什么



package main

import (
    "context"
    "crypto/tls"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"
    "os"
    "os/signal"
    "strings"
    "syscall"

    v1admission "k8s.io/api/admission/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/runtime/serializer"

    appv1 "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/klog"
)

type patch struct {
    Op    string            `json:"op"`
    Path  string            `json:"path"`
    Value map[string]string `json:"value"`
}

func serve(w http.ResponseWriter, r *http.Request) {

    var body []byte
    if data, err := ioutil.ReadAll(r.Body); err == nil {
        body = data
    }
    klog.Infof(fmt.Sprintf("receive request: %v....", string(body)[:130]))
    if len(body) == 0 {
        klog.Error(fmt.Sprintf("admission request body is empty"))
        http.Error(w, fmt.Errorf("admission request body is empty").Error(), http.StatusBadRequest)
        return
    }
    var admission v1admission.AdmissionReview
    codefc := serializer.NewCodecFactory(runtime.NewScheme())
    decoder := codefc.UniversalDeserializer()
    _, _, err := decoder.Decode(body, nil, &admission)

    if err != nil {
        msg := fmt.Sprintf("Request could not be decoded: %v", err)
        klog.Error(msg)
        http.Error(w, msg, http.StatusBadRequest)
        return
    }

    if admission.Request == nil {
        klog.Error(fmt.Sprintf("admission review can't be used: Request field is nil"))
        http.Error(w, fmt.Errorf("admission review can't be used: Request field is nil").Error(), http.StatusBadRequest)
        return
    }

    switch strings.Split(r.RequestURI, "?")[0] {
    case "/mutate":
        req := admission.Request
        var admissionResp v1admission.AdmissionReview
        admissionResp.APIVersion = admission.APIVersion
        admissionResp.Kind = admission.Kind
        klog.Infof("AdmissionReview for Kind=%v, Namespace=%v Name=%v UID=%v Operation=%v",
            req.Kind.Kind, req.Namespace, req.Name, req.UID, req.Operation)
        switch req.Kind.Kind {
        case "Deployment":
            var (
                respstr []byte
                err     error
                deploy  appv1.Deployment
            )
            if err = json.Unmarshal(req.Object.Raw, &deploy); err != nil {
                respStructure := v1admission.AdmissionResponse{Result: &metav1.Status{
                    Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
                    Code:    http.StatusInternalServerError,
                }}
                klog.Error(fmt.Sprintf("could not unmarshal resouces review request: %v", err))
                if respstr, err = json.Marshal(respStructure); err != nil {
                    klog.Error(fmt.Errorf("could not unmarshal resouces review response: %v", err))
                    http.Error(w, fmt.Errorf("could not unmarshal resouces review response: %v", err).Error(), http.StatusInternalServerError)
                    return
                }
                http.Error(w, string(respstr), http.StatusBadRequest)
                return
            }

            current_annotations := deploy.GetAnnotations()
            pl := []patch{}
            for k, v := range current_annotations {
                pl = append(pl, patch{
                    Op:   "add",
                    Path: "/metadata/annotations",
                    Value: map[string]string{
                        k: v,
                    },
                })
            }
            pl = append(pl, patch{
                Op:   "add",
                Path: "/metadata/annotations",
                Value: map[string]string{
                    deploy.Name + "/Allow": "true",
                },
            })

            annotationbyte, err := json.Marshal(pl)

            if err != nil {
                http.Error(w, err.Error(), http.StatusInternalServerError)
                return
            }
            respStructure := &v1admission.AdmissionResponse{
                UID:     req.UID,
                Allowed: true,
                Patch:   annotationbyte,
                PatchType: func() *v1admission.PatchType {
                    t := v1admission.PatchTypeJSONPatch
                    return &t
                }(),
                Result: &metav1.Status{
                    Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
                    Code:    http.StatusOK,
                },
            }
            admissionResp.Response = respStructure

            klog.Infof("sending response: %s....", admissionResp.Response.String()[:130])
            respByte, err := json.Marshal(admissionResp)
            if err != nil {
                klog.Errorf("Can't encode response messages: %v", err)
                http.Error(w, err.Error(), http.StatusInternalServerError)
            }
            klog.Infof("prepare to write response...")
            w.Header().Set("Content-Type", "application/json")
            if _, err := w.Write(respByte); err != nil {
                klog.Errorf("Can't write response: %v", err)
                http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
            }

        default:
            klog.Error(fmt.Sprintf("unsupport resouces review request type"))
            http.Error(w, "unsupport resouces review request type", http.StatusBadRequest)
        }

    case "/validate":
        req := admission.Request
        var admissionResp v1admission.AdmissionReview
        admissionResp.APIVersion = admission.APIVersion
        admissionResp.Kind = admission.Kind
        klog.Infof("AdmissionReview for Kind=%v, Namespace=%v Name=%v UID=%v Operation=%v",
            req.Kind.Kind, req.Namespace, req.Name, req.UID, req.Operation)
        var (
            deploy  appv1.Deployment
            respstr []byte
        )
        switch req.Kind.Kind {
        case "Deployment":
            if err = json.Unmarshal(req.Object.Raw, &deploy); err != nil {
                respStructure := v1admission.AdmissionResponse{Result: &metav1.Status{
                    Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
                    Code:    http.StatusInternalServerError,
                }}
                klog.Error(fmt.Sprintf("could not unmarshal resouces review request: %v", err))
                if respstr, err = json.Marshal(respStructure); err != nil {
                    klog.Error(fmt.Errorf("could not unmarshal resouces review response: %v", err))
                    http.Error(w, fmt.Errorf("could not unmarshal resouces review response: %v", err).Error(), http.StatusInternalServerError)
                    return
                }
                http.Error(w, string(respstr), http.StatusBadRequest)
                return
            }
        }
        al := deploy.GetAnnotations()
        respStructure := v1admission.AdmissionResponse{
            UID: req.UID,
        }
        if al[fmt.Sprintf("%s/Allow", deploy.Name)] == "true" {
            respStructure.Allowed = true
            respStructure.Result = &metav1.Status{
                Code: http.StatusOK,
            }
        } else {
            respStructure.Allowed = false
            respStructure.Result = &metav1.Status{
                Code: http.StatusForbidden,
                Reason: func() metav1.StatusReason {
                    return metav1.StatusReasonForbidden
                }(),
                Message: fmt.Sprintf("the resource %s couldn't to allow entry.", deploy.Kind),
            }
        }

        admissionResp.Response = &respStructure

        klog.Infof("sending response: %s....", admissionResp.Response.String()[:130])
        respByte, err := json.Marshal(admissionResp)
        if err != nil {
            klog.Errorf("Can't encode response messages: %v", err)
            http.Error(w, err.Error(), http.StatusInternalServerError)
        }
        klog.Infof("prepare to write response...")
        w.Header().Set("Content-Type", "application/json")
        if _, err := w.Write(respByte); err != nil {
            klog.Errorf("Can't write response: %v", err)
            http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
        }
    }
}

func main() {
    var (
        cert, key string
    )

    if cert = os.Getenv("TLS_CERT"); len(cert) == 0 {
        cert = "./tls/tls.crt"
    }

    if key = os.Getenv("TLS_KEY"); len(key) == 0 {
        key = "./tls/tls.key"
    }

    ca, err := tls.LoadX509KeyPair(cert, key)
    if err != nil {
        klog.Error(err.Error())
        return
    }

    server := &http.Server{
        Addr: ":81",
        TLSConfig: &tls.Config{
            Certificates: []tls.Certificate{
                ca,
            },
        },
    }

    httpserver := http.NewServeMux()

    httpserver.HandleFunc("/validate", serve)
    httpserver.HandleFunc("/mutate", serve)
    httpserver.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
        klog.Info(fmt.Sprintf("%s %s", r.RequestURI, "pong"))
        fmt.Fprint(w, "pong")
    })
    server.Handler = httpserver

    go func() {
        if err := server.ListenAndServeTLS("", ""); err != nil {
            klog.Errorf("Failed to listen and serve webhook server: %v", err)
        }
    }()

    klog.Info("starting serve.")
    signalChan := make(chan os.Signal, 1)
    signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
    <-signalChan

    klog.Infof("Got shut signal, shutting...")
    if err := server.Shutdown(context.Background()); err != nil {
        klog.Errorf("HTTP server Shutdown: %v", err)
    }
}

对应的Dockerfile



FROM golang:alpine AS builder
MAINTAINER cylon
WORKDIR /admission
COPY ./ /admission
ENV GOPROXY https://goproxy.cn,direct
RUN \
    sed -i 's/dl-cdn.alpinelinux.org/mirrors.ustc.edu.cn/g' /etc/apk/repositories && \
    apk add upx  && \
    GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags "-s -w" -o webhook main.go && \
    upx -1 webhook && \
    chmod +x webhook

FROM alpine AS runner
WORKDIR /go/admission
COPY --from=builder /admission/webhook .
VOLUME ["/admission"]

集群内部部署所需的资源清单



apiVersion: v1
kind: Service
metadata:
  name: admission-webhook
  labels:
    app: admission-webhook
spec:
  ports:
    - port: 81
      targetPort: 81
  selector:
    app: simple-webhook
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: simple-webhook
  name: simple-webhook
spec:
  replicas: 1
  selector:
    matchLabels:
      app: simple-webhook
  template:
    metadata:
      labels:
        app: simple-webhook
    spec:
      containers:
        - image: cylonchau/simple-webhook:v0.0.2
          imagePullPolicy: IfNotPresent
          name: webhook
          command: ["./webhook"]
          env:
            - name: "TLS_CERT"
              value: "./tls/tls.crt"
            - name: "TLS_KEY"
              value: "./tls/tls.key"
            - name: NS_NAME
              valueFrom:
                fieldRef:
                  apiVersion: v1
                  fieldPath: metadata.namespace
          ports:
            - containerPort: 81
          volumeMounts:
            - name: tlsdir
              mountPath: /go/admission/tls
              readOnly: true
      volumes:
        - name: tlsdir
          secret:
            secretName: webhook
---
apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: "pod-policy.example.com"
webhooks:
  - name: "pod-policy.example.com"
    rules:
      - apiGroups:   ["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
        apiVersions: ["v1"] # 拦截资源的版本
        operations:  ["CREATE"] # 什么请求下拦截
        resources:   ["deployments"]  # 拦截什么资源
        scope:       "Namespaced" # 生效的范围，cluster还是namespace "*"表示没有范围限制。
    clientConfig: # 我们部署的webhook服务，
      url: "https://10.0.0.1:81/mutate"
#      service: # service是在cluster-in模式下
#        namespace: "default"
#        name: "admission-webhook"
#        port: 81 # 服务的端口
#        path: "/mutate" # path是对应用于验证的接口
      # caBundle是提供给 admission webhook CA证书
      caBundle: Put you CA (base64 encode) in here
    admissionReviewVersions: ["v1"]
    sideEffects: None
    timeoutSeconds: 5 # 1-30s直接，表示请求api的超时时间
---
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: "valipod-policy.example.com"
webhooks:
- name: "valipod-policy.example.com"
  rules:
    - apiGroups:   ["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
      apiVersions: ["v1"] # 拦截资源的版本
      operations:  ["CREATE"] # 什么请求下拦截
      resources:   ["deployments"]  # 拦截什么资源
      scope:       "Namespaced" # 生效的范围，cluster还是namespace "*"表示没有范围限制。
  clientConfig: # 我们部署的webhook服务，
    #      service: # service是在cluster-in模式下
    #        namespace: "default"
    #        name: "admission-webhook"
    #        port: 81 # 服务的端口
    #        path: "/mutate" # path是对应用于验证的接口
    # caBundle是提供给 admission webhook CA证书
    caBundle: Put you CA (base64 encode) in here
  admissionReviewVersions: ["v1"]
  sideEffects: None
  timeoutSeconds: 5 # 1-30s直接，表示请求api的超时时间

这里需要主义的问题

证书问题

如果需要 cluster-in ，那么则需要对对应webhookconfig资源配置 service ；如果使用的是外部部署，则需要配置对应访问地址，如："https://xxxx:port/method"

这两种方式的证书均需要对应的 subjectAltName ，cluster-in 模式需要对应service名称，如，至少包含serviceName.NS.svc 这一个域名。

下面就是证书类问题的错误



Failed calling webhook, failing closed pod-policy.example.com: failed calling webhook "pod-policy.example.com": Post https://admission-webhook.default.svc:81/mutate?timeout=5s: x509: certificate signed by unknown authority (possibly because of "crypto/rsa: verification error" while trying to verify candidate authority certificate "admission-webhook-ca")

相应信息问题

上面我们了解到的APIServer是去发出 v1admission.AdmissionReview 也就是 Request 和 Response类型的，所以，为了更清晰的表示出问题所在，需要对响应格式中的 Reason 与 Message 配置，这也就是我们在客户端看到的报错信息。



&metav1.Status{
    Code: http.StatusForbidden,
    Reason: func() metav1.StatusReason {
        return metav1.StatusReasonForbidden
    }(),
    Message: fmt.Sprintf("the resource %s couldn't to allow entry.", deploy.Kind),
}

通过上面的设置用户可以看到下列错误



$ kubectl apply -f nginx.yaml 
Error from server (Forbidden): error when creating "nginx.yaml": admission webhook "valipod-policy.example.com" denied the request: the resource Deployment couldn't to allow entry.

注：必须的参数还包含，UID，allowed，这两个是必须的，上面阐述的只是对用户友好的提示信息

下面的报错就是对相应格式设置错误



Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: failed calling webhook "pod-policy.example.com": the server rejected our request for an unknown reason

相应信息版本问题

相应信息也需要指定一个版本，这个与请求来的结构中拿即可



admissionResp.APIVersion = admission.APIVersion
admissionResp.Kind = admission.Kind

下面是没有为对应相应信息配置对应KV的值出现的报错



Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: failed calling webhook "pod-policy.example.com": expected webhook response of admission.k8s.io/v1, Kind=AdmissionReview, got /, Kind=

关于patch

kubernetes中patch使用的是特定的规范，如 jsonpatch

kubernetes当前唯一支持的 patchType 是 JSONPatch。有关更多详细信息，请参见 JSON patch

对于 jsonpatch 是一个固定的类型，在go中必须定义其结构体
{
  "op": "add", // 做什么操作
  "path": "/spec/replicas", // 操作的路径
  "value": 3 // 对应添加的key value
}

下面就是字符串类型设置为布尔型产生的报错



Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: v1.Deployment.ObjectMeta: v1.ObjectMeta.Annotations: ReadString: expects " or n, but found t, error found in #10 byte of ...|t/Allow":true},"crea|..., bigger context ...|tadata":{"annotations":{"nginx-deployment/Allow":true},"creationTimestamp":null,"managedFields":[{"m|..

准备证书

Ubuntu



touch ./demoCAindex.txt
touch ./demoCA/serial 
touch ./demoCA/crlnumber
echo 01 > ./demoCA/serial
mkdir ./demoCA/newcerts

openssl genrsa -out cakey.pem 2048

openssl req -new \
    -x509 \
    -key cakey.pem \
    -out cacert.pem \
    -days 3650 \
    -subj "/CN=admission webhook ca"

openssl genrsa -out tls.key 2048

openssl req -new \
    -key tls.key \
    -subj "/CN=admission webhook client" \
    -reqexts webhook \
    -config <(cat /etc/ssl/openssl.cnf \
    <(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,  IP:10.0.0.4")) \
    -out tls.csr

sed -i 's/= match/= optional/g' /etc/ssl/openssl.cnf

openssl ca \
    -in tls.csr \
    -cert cacert.pem \
    -keyfile cakey.pem \
    -out tls.crt \
    -days 300 \
    -extensions webhook \
    -extfile <(cat /etc/ssl/openssl.cnf \
    <(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,  IP:10.0.0.4"))

CentOS



touch /etc/pki/CA/index.txt
touch /etc/pki/CA/serial # 下一个要颁发的编号 16进制
touch /etc/pki/CA/crlnumber
echo 01 > /etc/pki/CA/serial

openssl req -new \
    -x509 \
    -key cakey.pem \
    -out cacert.pem \
    -days 3650 \
    -subj "/CN=admission webhook ca"

openssl genrsa -out tls.key 2048

openssl req -new \
    -key tls.key \
    -subj "/CN=admission webhook client" \
    -reqexts webhook \
    -config <(cat /etc/pki/tls/openssl.cnf \
    <(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,  IP:10.0.0.4")) \
    -out tls.csr

sed -i 's/= match/= optional/g' /etc/ssl/openssl.cnf

openssl ca \
    -in tls.csr \
    -cert cacert.pem \
    -keyfile cakey.pem \
    -out tls.crt \
    -days 300 \
    -extensions webhook \
    -extfile <(cat /etc/pki/tls/openssl.cnf \
    <(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,  IP:10.0.0.4"))

通过部署测试结果

可以看到我们自己注入的 annotation nginx-deployment/Allow: true，在该示例中，仅为演示过程，而不是真的策略，实际环境中可以根据情况进行定制自己的策略。

结果可以看出，当在 mutating 中不通过，即缺少对应的 annotation 标签 , 则 validating 会不允许准入



$ kubectl describe deploy nginx-deployment
Name:                   nginx-deployment
Namespace:              default
CreationTimestamp:      Mon, 11 Jul 2022 20:25:16 +0800
Labels:                 <none>
Annotations:            deployment.kubernetes.io/revision: 1
                        nginx-deployment/Allow: true
Selector:               app=nginx
Replicas:               1 desired | 1 updated | 1 total | 1 available | 0 unavailable
StrategyType:           RollingUpdate
MinReadySeconds:        0
RollingUpdateStrategy:  25% max unavailable, 25% max surge
Pod Template:
  Labels:  app=nginx
  Containers:
   nginx:
    Image:        nginx:1.14.2

Reference

extensible admission controllers

K8S client-go Patch example

admission controllers response

a guide to kubernetes admission controllers

kube-scheduler的调度上下文

cylon — Sun, 31 Jul 2022 14:35:45 +0000

Scheduler结构

Scheduler 是整个 kube-scheduler 的一个 structure，提供了 kube-scheduler 运行所需的组件。

type Scheduler struct {
    // Cache是一个抽象，会缓存pod的信息，作为scheduler进行查找，操作是基于Pod进行增加
    Cache internalcache.Cache
    // Extenders 算是调度框架中提供的调度插件，会影响kubernetes中的调度策略
    Extenders []framework.Extender

    // NextPod 作为一个函数提供，会阻塞获取下一个ke'diao'du
    NextPod func() *framework.QueuedPodInfo

    // Error is called if there is an error. It is passed the pod in
    // question, and the error
    Error func(*framework.QueuedPodInfo, error)

    // SchedulePod 尝试将给出的pod调度到Node。
    SchedulePod func(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (ScheduleResult, error)

    // 关闭scheduler的信号
    StopEverything <-chan struct{}

    // SchedulingQueue保存要调度的Pod
    SchedulingQueue internalqueue.SchedulingQueue

    // Profiles中是多个调度框架
    Profiles profile.Map
    client clientset.Interface
    nodeInfoSnapshot *internalcache.Snapshot
    percentageOfNodesToScore int32
    nextStartNodeIndex int
}

作为实际执行的两个核心，SchedulingQueue ，与 scheduleOne 将会分析到这两个

SchedulingQueue

在知道 kube-scheduler 初始化过程后，需要对 kube-scheduler 的整个 structure 和 workflow 进行分析

在 Run 中，运行的是一个 SchedulingQueue 与一个 scheduleOne ，从结构上看是属于 Scheduler

func (sched *Scheduler) Run(ctx context.Context) {
    sched.SchedulingQueue.Run()

    // We need to start scheduleOne loop in a dedicated goroutine,
    // because scheduleOne function hangs on getting the next item
    // from the SchedulingQueue.
    // If there are no new pods to schedule, it will be hanging there
    // and if done in this goroutine it will be blocking closing
    // SchedulingQueue, in effect causing a deadlock on shutdown.
    go wait.UntilWithContext(ctx, sched.scheduleOne, 0)

    <-ctx.Done()
    sched.SchedulingQueue.Close()
}

SchedulingQueue 是一个队列的抽象，用于存储等待调度的Pod。该接口遵循类似于 cache.FIFO 和 cache.Heap 的模式。

type SchedulingQueue interface {
    framework.PodNominator
    Add(pod *v1.Pod) error
    // Activate moves the given pods to activeQ iff they're in unschedulablePods or backoffQ.
    // The passed-in pods are originally compiled from plugins that want to activate Pods,
    // by injecting the pods through a reserved CycleState struct (PodsToActivate).
    Activate(pods map[string]*v1.Pod)
    // 将不可调度的Pod重入到队列中
    AddUnschedulableIfNotPresent(pod *framework.QueuedPodInfo, podSchedulingCycle int64) error
    // SchedulingCycle returns the current number of scheduling cycle which is
    // cached by scheduling queue. Normally, incrementing this number whenever
    // a pod is popped (e.g. called Pop()) is enough.
    SchedulingCycle() int64
    // Pop会弹出一个pod，并从head优先级队列中删除
    Pop() (*framework.QueuedPodInfo, error)
    Update(oldPod, newPod *v1.Pod) error
    Delete(pod *v1.Pod) error
    MoveAllToActiveOrBackoffQueue(event framework.ClusterEvent, preCheck PreEnqueueCheck)
    AssignedPodAdded(pod *v1.Pod)
    AssignedPodUpdated(pod *v1.Pod)
    PendingPods() []*v1.Pod
    // Close closes the SchedulingQueue so that the goroutine which is
    // waiting to pop items can exit gracefully.
    Close()
    // Run starts the goroutines managing the queue.
    Run()
}

而 PriorityQueue 是 SchedulingQueue 的实现，该部分的核心构成是两个子队列与一个数据结构，即 activeQ、backoffQ 和 unschedulablePods

activeQ：是一个 heap 类型的优先级队列，是 sheduler 从中获得优先级最高的Pod进行调度
backoffQ：也是一个 heap 类型的优先级队列，存放的是不可调度的Pod
unschedulablePods：保存确定不可被调度的Pod

type SchedulingQueue interface {
    framework.PodNominator
    Add(pod *v1.Pod) error
    // Activate moves the given pods to activeQ iff they're in unschedulablePods or backoffQ.
    // The passed-in pods are originally compiled from plugins that want to activate Pods,
    // by injecting the pods through a reserved CycleState struct (PodsToActivate).
    Activate(pods map[string]*v1.Pod)
    // AddUnschedulableIfNotPresent adds an unschedulable pod back to scheduling queue.
    // The podSchedulingCycle represents the current scheduling cycle number which can be
    // returned by calling SchedulingCycle().
    AddUnschedulableIfNotPresent(pod *framework.QueuedPodInfo, podSchedulingCycle int64) error
    // SchedulingCycle returns the current number of scheduling cycle which is
    // cached by scheduling queue. Normally, incrementing this number whenever
    // a pod is popped (e.g. called Pop()) is enough.
    SchedulingCycle() int64
    // Pop removes the head of the queue and returns it. It blocks if the
    // queue is empty and waits until a new item is added to the queue.
    Pop() (*framework.QueuedPodInfo, error)
    Update(oldPod, newPod *v1.Pod) error
    Delete(pod *v1.Pod) error
    MoveAllToActiveOrBackoffQueue(event framework.ClusterEvent, preCheck PreEnqueueCheck)
    AssignedPodAdded(pod *v1.Pod)
    AssignedPodUpdated(pod *v1.Pod)
    PendingPods() []*v1.Pod
    // Close closes the SchedulingQueue so that the goroutine which is
    // waiting to pop items can exit gracefully.
    Close()
    // Run starts the goroutines managing the queue.
    Run()
}

在New scheduler 时可以看到会初始化这个queue

podQueue := internalqueue.NewSchedulingQueue(
    // 实现pod对比的一个函数即less
    profiles[options.profiles[0].SchedulerName].QueueSortFunc(),
    informerFactory,
    internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),
    internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),
    internalqueue.WithPodNominator(nominator),
    internalqueue.WithClusterEventMap(clusterEventMap),
    internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),
)

而 NewSchedulingQueue 则是初始化这个 PriorityQueue

// NewSchedulingQueue initializes a priority queue as a new scheduling queue.
func NewSchedulingQueue(
    lessFn framework.LessFunc,
    informerFactory informers.SharedInformerFactory,
    opts ...Option) SchedulingQueue {
    return NewPriorityQueue(lessFn, informerFactory, opts...)
}

// NewPriorityQueue creates a PriorityQueue object.
func NewPriorityQueue(
    lessFn framework.LessFunc,
    informerFactory informers.SharedInformerFactory,
    opts ...Option,
) *PriorityQueue {
    options := defaultPriorityQueueOptions
    for _, opt := range opts {
        opt(&options)
    }
    // 这个就是 less函数，作为打分的一部分
    comp := func(podInfo1, podInfo2 interface{}) bool {
        pInfo1 := podInfo1.(*framework.QueuedPodInfo)
        pInfo2 := podInfo2.(*framework.QueuedPodInfo)
        return lessFn(pInfo1, pInfo2)
    }

    if options.podNominator == nil {
        options.podNominator = NewPodNominator(informerFactory.Core().V1().Pods().Lister())
    }

    pq := &PriorityQueue{
        PodNominator:                      options.podNominator,
        clock:                             options.clock,
        stop:                              make(chan struct{}),
        podInitialBackoffDuration:         options.podInitialBackoffDuration,
        podMaxBackoffDuration:             options.podMaxBackoffDuration,
        podMaxInUnschedulablePodsDuration: options.podMaxInUnschedulablePodsDuration,
        activeQ:                           heap.NewWithRecorder(podInfoKeyFunc, comp, metrics.NewActivePodsRecorder()),
        unschedulablePods:                 newUnschedulablePods(metrics.NewUnschedulablePodsRecorder()),
        moveRequestCycle:                  -1,
        clusterEventMap:                   options.clusterEventMap,
    }
    pq.cond.L = &pq.lock
    pq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, pq.podsCompareBackoffCompleted, metrics.NewBackoffPodsRecorder())
    pq.nsLister = informerFactory.Core().V1().Namespaces().Lister()

    return pq
}

了解了Queue的结构，就需要知道入队列与出队列是在哪里操作的。在初始化时，需要注册一个 addEventHandlerFuncs 这个时候，会注入三个动作函数，也就是controller中的概念；而在AddFunc中可以看到会入队列。

注入是对 Pod 的informer注入的，注入的函数 addPodToSchedulingQueue 就是入栈

Handler: cache.ResourceEventHandlerFuncs{
    AddFunc:    sched.addPodToSchedulingQueue,
    UpdateFunc: sched.updatePodInSchedulingQueue,
    DeleteFunc: sched.deletePodFromSchedulingQueue,
},

func (sched *Scheduler) addPodToSchedulingQueue(obj interface{}) {
    pod := obj.(*v1.Pod)
    klog.V(3).InfoS("Add event for unscheduled pod", "pod", klog.KObj(pod))
    if err := sched.SchedulingQueue.Add(pod); err != nil {
        utilruntime.HandleError(fmt.Errorf("unable to queue %T: %v", obj, err))
    }
}

而这个 SchedulingQueue 的实现就是 PriorityQueue ，而Add中则对 activeQ进行的操作

func (p *PriorityQueue) Add(pod *v1.Pod) error {
    p.lock.Lock()
    defer p.lock.Unlock()
    // 格式化入栈数据，包含podinfo，里会包含v1.Pod
    // 初始化的时间，创建的时间，以及不能被调度时的记录其plugin的名称
    pInfo := p.newQueuedPodInfo(pod)
    // 入栈
    if err := p.activeQ.Add(pInfo); err != nil {
        klog.ErrorS(err, "Error adding pod to the active queue", "pod", klog.KObj(pod))
        return err
    }
    if p.unschedulablePods.get(pod) != nil {
        klog.ErrorS(nil, "Error: pod is already in the unschedulable queue", "pod", klog.KObj(pod))
        p.unschedulablePods.delete(pod)
    }
    // Delete pod from backoffQ if it is backing off
    if err := p.podBackoffQ.Delete(pInfo); err == nil {
        klog.ErrorS(nil, "Error: pod is already in the podBackoff queue", "pod", klog.KObj(pod))
    }
    metrics.SchedulerQueueIncomingPods.WithLabelValues("active", PodAdd).Inc()
    p.PodNominator.AddNominatedPod(pInfo.PodInfo, nil)
    p.cond.Broadcast()

    return nil
}

在上面看 scheduler 结构时，可以看到有一个 nextPod的，nextPod就是从队列中弹出一个pod，这个在scheduler 时会传入 MakeNextPodFunc 就是这个 nextpod

func MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo {
    return func() *framework.QueuedPodInfo {
        podInfo, err := queue.Pop()
        if err == nil {
            klog.V(4).InfoS("About to try and schedule pod", "pod", klog.KObj(podInfo.Pod))
            for plugin := range podInfo.UnschedulablePlugins {
                metrics.UnschedulableReason(plugin, podInfo.Pod.Spec.SchedulerName).Dec()
            }
            return podInfo
        }
        klog.ErrorS(err, "Error while retrieving next pod from scheduling queue")
        return nil
    }
}

而这个 queue.Pop() 对应的就是 PriorityQueue 的 Pop() ，在这里会将作为 activeQ 的消费端

func (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error) {
   p.lock.Lock()
   defer p.lock.Unlock()
   for p.activeQ.Len() == 0 {
      // When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
      // When Close() is called, the p.closed is set and the condition is broadcast,
      // which causes this loop to continue and return from the Pop().
      if p.closed {
         return nil, fmt.Errorf(queueClosed)
      }
      p.cond.Wait()
   }
   obj, err := p.activeQ.Pop()
   if err != nil {
      return nil, err
   }
   pInfo := obj.(*framework.QueuedPodInfo)
   pInfo.Attempts++
   p.schedulingCycle++
   return pInfo, nil
}

在上面入口部分也看到了，scheduleOne 和 scheduler，scheduleOne 就是去消费一个Pod，他会调用 NextPod，NextPod就是在初始化传入的 MakeNextPodFunc ，至此回到对应的 Pop来做消费。

schedulerOne是为一个Pod做调度的流程。

func (sched *Scheduler) scheduleOne(ctx context.Context) {
    podInfo := sched.NextPod()
    // pod could be nil when schedulerQueue is closed
    if podInfo == nil || podInfo.Pod == nil {
        return
    }
    pod := podInfo.Pod
    fwk, err := sched.frameworkForPod(pod)
    if err != nil {
        // This shouldn't happen, because we only accept for scheduling the pods
        // which specify a scheduler name that matches one of the profiles.
        klog.ErrorS(err, "Error occurred")
        return
    }
    if sched.skipPodSchedule(fwk, pod) {
        return
    }
...

调度上下文

图1：Pod的调度上下文
Source：https://kubernetes.io/docs/concepts/scheduling-eviction/scheduling-framework

当了解了scheduler结构后，下面分析下调度上下文的过程。看看扩展点是怎么工作的。这个时候又需要提到官网的调度上下文的图。

调度框架 ^[2]

调度框架 (scheduling framework SF ) 是kubernetes为 scheduler设计的一个pluggable的架构。SF 将scheduler设计为 Plugin 式的 API，API将上一章中提到的一些列调度策略实现为 Plugin。

在 SF 中，定义了一些扩展点（extension points EP ），而被实现为Plugin的调度程序将被注册在一个或多个 EP 中，换句话来说，在这些 EP 的执行过程中如果注册在多个 EP 中，将会在多个 EP 被调用。

每次调度都分为两个阶段，调度周期（Scheduling Cycel）与绑定周期（Binding Cycle）。

SC 表示为，为Pod选择一个节点；SC 是串行运行的。
BC 表示为，将 SC 决策结果应用于集群中；BC 可以同时运行。

调度周期与绑定周期结合一起，被称为调度上下文 （Scheduling Context）,下图则是调度上下文的工作流

注：如果决策结果为Pod的调度结果无可用节点，或存在内部错误，则中止 SC 或 BC。Pod将重入队列重试

扩展点 ^[3]

扩展点（Extension points）是指在调度上下文中的每个可扩展API，通过图提现为[图1]。其中 Filter 相当于 Predicate 而 Scoring 相当于 Priority。

对于调度阶段会通过以下扩展点：

Sort：该插件提供了排序功能，用于对在调度队列中待处理 Pod 进行排序。一次只能启用一个队列排序。
preFilter：该插件用于在过滤之前预处理或检查 Pod 或集群的相关信息。这里会终止调度
filter：该插件相当于调度上下文中的 Predicates，用于排除不能运行 Pod 的节点。Filter 会按配置的顺序进行调用。如果有一个filter将节点标记位不可用，则将 Pod 标记为不可调度（即不会向下执行）。
postFilter：当没有为 pod 找到FN时，该插件会按照配置的顺序进行调用。如果任何postFilter插件将 Pod 标记为schedulable，则不会调用其余插件。即 filter 成功后不会进行这步骤
preScore：可用于进行预Score工作（通知性的扩展点）。
score：该插件为每个通过 filter 阶段的Node提供打分服务。然后Scheduler将选择具有最高加权分数总和的Node。
reserve：因为绑定事件时异步发生的，该插件是为了避免Pod在绑定到节点前时，调度到新的Pod，使节点使用资源超过可用资源情况。如果后续阶段发生错误或失败，将触发 UnReserve 回滚（通知性扩展点）。这也是作为调度周期中最后一个状态，要么成功到 postBind ，要么失败触发 UnReserve。
permit：该插件可以阻止或延迟 Pod 的绑定，一般情况下这步骤会做三件事：
- appove ：调度器继续绑定过程
- Deny：如果任何一个Premit拒绝了Pod与节点的绑定，那么将触发 UnReserve ，并重入队列
- Wait：如果 Permit 插件返回 Wait，该 Pod 将保留在内部 Wait Pod 列表中，直到被 Appove。如果发生超时，wait 变为 deny ，将Pod放回至调度队列中，并触发 Unreserve 回滚。
preBind：该插件用于在 bind Pod 之前执行所需的前置工作。如，preBind 可能会提供一个网络卷并将其挂载到目标节点上。如果在该步骤中的任意插件返回错误，则Pod 将被 deny 并放置到调度队列中。
bind：在所有的 preBind 完成后，该插件将用于将Pod绑定到Node，并按顺序调用绑定该步骤的插件。如果有一个插件处理了这个事件，那么则忽略其余所有插件。
postBind：该插件在绑定 Pod 后调用，可用于清理相关资源（通知性的扩展点）。
multiPoint：这是一个仅配置字段，允许同时为所有适用的扩展点启用或禁用插件。

而 scheduler 对于调度上下文在代码中的实现就是 scheduleOne ，下面就是看这个调度上下文

Sort

Sort 插件提供了排序功能，用于对在调度队列中待处理 Pod 进行排序。一次只能启用一个队列排序。

在进入 scheduleOne 后，NextPod 从 activeQ 中队列中得到一个Pod，然后的 frameworkForPod 会做打分的动作就是调度上下文的第一个扩展点 sort

func (sched *Scheduler) scheduleOne(ctx context.Context) {
    podInfo := sched.NextPod()
    // pod could be nil when schedulerQueue is closed
    if podInfo == nil || podInfo.Pod == nil {
        return
    }
    pod := podInfo.Pod
    fwk, err := sched.frameworkForPod(pod)
...

func (sched *Scheduler) frameworkForPod(pod *v1.Pod) (framework.Framework, error) {
    // 获取指定的profile
    fwk, ok := sched.Profiles[pod.Spec.SchedulerName]
    if !ok {
        return nil, fmt.Errorf("profile not found for scheduler name %q", pod.Spec.SchedulerName)
    }
    return fwk, nil
}

回顾，因为在New scheduler时会初始化这个 sort 函数

podQueue := internalqueue.NewSchedulingQueue(
    profiles[options.profiles[0].SchedulerName].QueueSortFunc(),
    informerFactory,
    internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),
    internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),
    internalqueue.WithPodNominator(nominator),
    internalqueue.WithClusterEventMap(clusterEventMap),
    internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),
)

preFilter

preFilter作为第一个扩展点，是用于在过滤之前预处理或检查 Pod 或集群的相关信息。这里会终止调度

func (sched *Scheduler) scheduleOne(ctx context.Context) {
    podInfo := sched.NextPod()
    // pod could be nil when schedulerQueue is closed
    if podInfo == nil || podInfo.Pod == nil {
        return
    }
    pod := podInfo.Pod
    fwk, err := sched.frameworkForPod(pod)
    if err != nil {
        // This shouldn't happen, because we only accept for scheduling the pods
        // which specify a scheduler name that matches one of the profiles.
        klog.ErrorS(err, "Error occurred")
        return
    }
    if sched.skipPodSchedule(fwk, pod) {
        return
    }

    klog.V(3).InfoS("Attempting to schedule pod", "pod", klog.KObj(pod))

    // Synchronously attempt to find a fit for the pod.
    start := time.Now()
    state := framework.NewCycleState()
    state.SetRecordPluginMetrics(rand.Intn(100) < pluginMetricsSamplePercent)
    // Initialize an empty podsToActivate struct, which will be filled up by plugins or stay empty.
    podsToActivate := framework.NewPodsToActivate()
    state.Write(framework.PodsToActivateKey, podsToActivate)

    schedulingCycleCtx, cancel := context.WithCancel(ctx)
    defer cancel()
    // 这里将进入prefilter
    scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)

schedulePod 尝试将给定的 pod 调度到节点列表中的节点之一。如果成功，它将返回节点的名称。

func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
    trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
    defer trace.LogIfLong(100 * time.Millisecond)
    // 用于将cache更新为当前内容
    if err := sched.Cache.UpdateSnapshot(sched.nodeInfoSnapshot); err != nil {
        return result, err
    }
    trace.Step("Snapshotting scheduler cache and node infos done")

    if sched.nodeInfoSnapshot.NumNodes() == 0 {
        return result, ErrNoNodesAvailable
    }
    // 找到一个合适的pod时，会执行扩展点
    feasibleNodes, diagnosis, err := sched.findNodesThatFitPod(ctx, fwk, state, pod)

    ...

findNodesThatFitPod 会执行对应的过滤插件来找到最适合的Node，包括备注，以及方法名都可以看到，这里运行的插件😁😁，后面会分析算法内容，只对workflow学习。

func (sched *Scheduler) findNodesThatFitPod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.Diagnosis, error) {
    diagnosis := framework.Diagnosis{
        NodeToStatusMap:      make(framework.NodeToStatusMap),
        UnschedulablePlugins: sets.NewString(),
    }

    // Run "prefilter" plugins.
    preRes, s := fwk.RunPreFilterPlugins(ctx, state, pod)
    allNodes, err := sched.nodeInfoSnapshot.NodeInfos().List()
    if err != nil {
        return nil, diagnosis, err
    }
    if !s.IsSuccess() {
        if !s.IsUnschedulable() {
            return nil, diagnosis, s.AsError()
        }
        // All nodes will have the same status. Some non trivial refactoring is
        // needed to avoid this copy.
        for _, n := range allNodes {
            diagnosis.NodeToStatusMap[n.Node().Name] = s
        }
        // Status satisfying IsUnschedulable() gets injected into diagnosis.UnschedulablePlugins.
        if s.FailedPlugin() != "" {
            diagnosis.UnschedulablePlugins.Insert(s.FailedPlugin())
        }
        return nil, diagnosis, nil
    }

    // "NominatedNodeName" can potentially be set in a previous scheduling cycle as a result of preemption.
    // This node is likely the only candidate that will fit the pod, and hence we try it first before iterating over all nodes.
    if len(pod.Status.NominatedNodeName) > 0 {
        feasibleNodes, err := sched.evaluateNominatedNode(ctx, pod, fwk, state, diagnosis)
        if err != nil {
            klog.ErrorS(err, "Evaluation failed on nominated node", "pod", klog.KObj(pod), "node", pod.Status.NominatedNodeName)
        }
        // Nominated node passes all the filters, scheduler is good to assign this node to the pod.
        if len(feasibleNodes) != 0 {
            return feasibleNodes, diagnosis, nil
        }
    }

    nodes := allNodes
    if !preRes.AllNodes() {
        nodes = make([]*framework.NodeInfo, 0, len(preRes.NodeNames))
        for n := range preRes.NodeNames {
            nInfo, err := sched.nodeInfoSnapshot.NodeInfos().Get(n)
            if err != nil {
                return nil, diagnosis, err
            }
            nodes = append(nodes, nInfo)
        }
    }
    feasibleNodes, err := sched.findNodesThatPassFilters(ctx, fwk, state, pod, diagnosis, nodes)
    if err != nil {
        return nil, diagnosis, err
    }

    feasibleNodes, err = findNodesThatPassExtenders(sched.Extenders, pod, feasibleNodes, diagnosis.NodeToStatusMap)
    if err != nil {
        return nil, diagnosis, err
    }
    return feasibleNodes, diagnosis, nil
}

filter

filter插件相当于调度上下文中的 Predicates，用于排除不能运行 Pod 的节点。Filter 会按配置的顺序进行调用。如果有一个filter将节点标记位不可用，则将 Pod 标记为不可调度（即不会向下执行）。

对于代码中来讲，filter还是处于 findNodesThatFitPod 函数中，findNodesThatPassFilters 就是获取到 FN，即可行节点，而这个过程就是 filter 扩展点

func (sched *Scheduler) findNodesThatFitPod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.Diagnosis, error) {
    ...

    feasibleNodes, err := sched.findNodesThatPassFilters(ctx, fwk, state, pod, diagnosis, nodes)
    if err != nil {
        return nil, diagnosis, err
    }

    feasibleNodes, err = findNodesThatPassExtenders(sched.Extenders, pod, feasibleNodes, diagnosis.NodeToStatusMap)
    if err != nil {
        return nil, diagnosis, err
    }
    return feasibleNodes, diagnosis, nil
}

Postfilter

当没有为 pod 找到FN时，该插件会按照配置的顺序进行调用。如果任何postFilter插件将 Pod 标记为schedulable，则不会调用其余插件。即 filter 成功后不会进行这步骤，那我们来验证下这里把😊

还是在 scheduleOne 中，当我们运行的 SchedulePod 完成后（成功或失败），这时会返回一个err，而 postfilter 会根据这个 err进行选择执行或不执行，符合官方给出的说法。

scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)
    if err != nil {
        // SchedulePod() may have failed because the pod would not fit on any host, so we try to
        // preempt, with the expectation that the next time the pod is tried for scheduling it
        // will fit due to the preemption. It is also possible that a different pod will schedule
        // into the resources that were preempted, but this is harmless.
        var nominatingInfo *framework.NominatingInfo
        if fitError, ok := err.(*framework.FitError); ok {
            if !fwk.HasPostFilterPlugins() {
                klog.V(3).InfoS("No PostFilter plugins are registered, so no preemption will be performed")
            } else {
                // Run PostFilter plugins to try to make the pod schedulable in a future scheduling cycle.
                result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap)
                if status.Code() == framework.Error {
                    klog.ErrorS(nil, "Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                } else {
                    fitError.Diagnosis.PostFilterMsg = status.Message()
                    klog.V(5).InfoS("Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                }
                if result != nil {
                    nominatingInfo = result.NominatingInfo
                }
            }
            // Pod did not fit anywhere, so it is counted as a failure. If preemption
            // succeeds, the pod should get counted as a success the next time we try to
            // schedule it. (hopefully)
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else if err == ErrNoNodesAvailable {
            nominatingInfo = clearNominatedNode
            // No nodes available is counted as unschedulable rather than an error.
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else {
            nominatingInfo = clearNominatedNode
            klog.ErrorS(err, "Error selecting node for pod", "pod", klog.KObj(pod))
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        }
        sched.handleSchedulingFailure(ctx, fwk, podInfo, err, v1.PodReasonUnschedulable, nominatingInfo)
        return
    }

PreScore,Score

可用于进行预Score工作，作为通知性的扩展点，会在在filter完之后直接会关联 preScore 插件进行继续工作，而不是返回，如果配置的这些插件有任何一个返回失败，则Pod将被拒绝。


func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
    trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
    defer trace.LogIfLong(100 * time.Millisecond)

    if err := sched.Cache.UpdateSnapshot(sched.nodeInfoSnapshot); err != nil {
        return result, err
    }
    trace.Step("Snapshotting scheduler cache and node infos done")

    if sched.nodeInfoSnapshot.NumNodes() == 0 {
        return result, ErrNoNodesAvailable
    }

    feasibleNodes, diagnosis, err := sched.findNodesThatFitPod(ctx, fwk, state, pod)
    if err != nil {
        return result, err
    }
    trace.Step("Computing predicates done")

    if len(feasibleNodes) == 0 {
        return result, &framework.FitError{
            Pod:         pod,
            NumAllNodes: sched.nodeInfoSnapshot.NumNodes(),
            Diagnosis:   diagnosis,
        }
    }

    // When only one node after predicate, just use it.
    if len(feasibleNodes) == 1 {
        return ScheduleResult{
            SuggestedHost:  feasibleNodes[0].Name,
            EvaluatedNodes: 1 + len(diagnosis.NodeToStatusMap),
            FeasibleNodes:  1,
        }, nil
    }
    // 这里会完成prescore，score
    priorityList, err := prioritizeNodes(ctx, sched.Extenders, fwk, state, pod, feasibleNodes)
    if err != nil {
        return result, err
    }

    host, err := selectHost(priorityList)
    trace.Step("Prioritizing done")

    return ScheduleResult{
        SuggestedHost:  host,
        EvaluatedNodes: len(feasibleNodes) + len(diagnosis.NodeToStatusMap),
        FeasibleNodes:  len(feasibleNodes),
    }, err
}

priorityNodes 会通过配置的插件给Node打分，并返回每个Node的分数，将每个插件打分结果计算总和获得Node的分数，最后获得节点的加权总分数。

func prioritizeNodes(
    ctx context.Context,
    extenders []framework.Extender,
    fwk framework.Framework,
    state *framework.CycleState,
    pod *v1.Pod,
    nodes []*v1.Node,
) (framework.NodeScoreList, error) {
    // If no priority configs are provided, then all nodes will have a score of one.
    // This is required to generate the priority list in the required format
    if len(extenders) == 0 && !fwk.HasScorePlugins() {
        result := make(framework.NodeScoreList, 0, len(nodes))
        for i := range nodes {
            result = append(result, framework.NodeScore{
                Name:  nodes[i].Name,
                Score: 1,
            })
        }
        return result, nil
    }

    // Run PreScore plugins.
    preScoreStatus := fwk.RunPreScorePlugins(ctx, state, pod, nodes)
    if !preScoreStatus.IsSuccess() {
        return nil, preScoreStatus.AsError()
    }

    // Run the Score plugins.
    scoresMap, scoreStatus := fwk.RunScorePlugins(ctx, state, pod, nodes)
    if !scoreStatus.IsSuccess() {
        return nil, scoreStatus.AsError()
    }

    // Additional details logged at level 10 if enabled.
    klogV := klog.V(10)
    if klogV.Enabled() {
        for plugin, nodeScoreList := range scoresMap {
            for _, nodeScore := range nodeScoreList {
                klogV.InfoS("Plugin scored node for pod", "pod", klog.KObj(pod), "plugin", plugin, "node", nodeScore.Name, "score", nodeScore.Score)
            }
        }
    }

    // Summarize all scores.
    result := make(framework.NodeScoreList, 0, len(nodes))

    for i := range nodes {
        result = append(result, framework.NodeScore{Name: nodes[i].Name, Score: 0})
        for j := range scoresMap {
            result[i].Score += scoresMap[j][i].Score
        }
    }

    if len(extenders) != 0 && nodes != nil {
        var mu sync.Mutex
        var wg sync.WaitGroup
        combinedScores := make(map[string]int64, len(nodes))
        for i := range extenders {
            if !extenders[i].IsInterested(pod) {
                continue
            }
            wg.Add(1)
            go func(extIndex int) {
                metrics.SchedulerGoroutines.WithLabelValues(metrics.PrioritizingExtender).Inc()
                defer func() {
                    metrics.SchedulerGoroutines.WithLabelValues(metrics.PrioritizingExtender).Dec()
                    wg.Done()
                }()
                prioritizedList, weight, err := extenders[extIndex].Prioritize(pod, nodes)
                if err != nil {
                    // Prioritization errors from extender can be ignored, let k8s/other extenders determine the priorities
                    klog.V(5).InfoS("Failed to run extender's priority function. No score given by this extender.", "error", err, "pod", klog.KObj(pod), "extender", extenders[extIndex].Name())
                    return
                }
                mu.Lock()
                for i := range *prioritizedList {
                    host, score := (*prioritizedList)[i].Host, (*prioritizedList)[i].Score
                    if klogV.Enabled() {
                        klogV.InfoS("Extender scored node for pod", "pod", klog.KObj(pod), "extender", extenders[extIndex].Name(), "node", host, "score", score)
                    }
                    combinedScores[host] += score * weight
                }
                mu.Unlock()
            }(i)
        }
        // wait for all go routines to finish
        wg.Wait()
        for i := range result {
            // MaxExtenderPriority may diverge from the max priority used in the scheduler and defined by MaxNodeScore,
            // therefore we need to scale the score returned by extenders to the score range used by the scheduler.
            result[i].Score += combinedScores[result[i].Name] * (framework.MaxNodeScore / extenderv1.MaxExtenderPriority)
        }
    }

    if klogV.Enabled() {
        for i := range result {
            klogV.InfoS("Calculated node's final score for pod", "pod", klog.KObj(pod), "node", result[i].Name, "score", result[i].Score)
        }
    }
    return result, nil
}

Reserve

Reserve 因为绑定事件时异步发生的，该插件是为了避免Pod在绑定到节点前时，调度到新的Pod，使节点使用资源超过可用资源情况。如果后续阶段发生错误或失败，将触发 UnReserve 回滚（通知性扩展点）。这也是作为调度周期中最后一个状态，要么成功到 postBind ，要么失败触发 UnReserve。

// Run the Reserve method of reserve plugins.
if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() { // 当处理不成功时
    metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
    // 触发 un-reserve 来清理相关Pod的状态
    fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if forgetErr := sched.Cache.ForgetPod(assumedPod); forgetErr != nil {
        klog.ErrorS(forgetErr, "Scheduler cache ForgetPod failed")
    }
    sched.handleSchedulingFailure(ctx, fwk, assumedPodInfo, sts.AsError(), SchedulerError, clearNominatedNode)
    return
}

permit

Permit 插件可以阻止或延迟 Pod 的绑定

    // Run "permit" plugins.
    runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if !runPermitStatus.IsWait() && !runPermitStatus.IsSuccess() {
        var reason string
        if runPermitStatus.IsUnschedulable() {
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
            reason = v1.PodReasonUnschedulable
        } else {
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
            reason = SchedulerError
        }
        // 只要其中一个插件返回的状态不是 success 或者 wait
        fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
        // 从cache中忘掉pod
        if forgetErr := sched.Cache.ForgetPod(assumedPod); forgetErr != nil {
            klog.ErrorS(forgetErr, "Scheduler cache ForgetPod failed")
        }
        sched.handleSchedulingFailure(ctx, fwk, assumedPodInfo, runPermitStatus.AsError(), reason, clearNominatedNode)
        return
    }

Binding Cycle

在选择好 FN 后则做一个假设绑定，并更新到cache中，接下来回去执行真正的bind操作，也就是 binding cycle

func (sched *Scheduler) scheduleOne(ctx context.Context) {
    ...
    ...
    // binding cycle 是一个异步的操作，这里表现就是go协程
    go func() {
        bindingCycleCtx, cancel := context.WithCancel(ctx)
        defer cancel()
        metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Inc()
        defer metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Dec()
        // 运行WaitOnPermit插件，如果失败则，unReserve回滚
        waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
        if !waitOnPermitStatus.IsSuccess() {
            var reason string
            if waitOnPermitStatus.IsUnschedulable() {
                metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
                reason = v1.PodReasonUnschedulable
            } else {
                metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
                reason = SchedulerError
            }
            // trigger un-reserve plugins to clean up state associated with the reserved Pod
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
            if forgetErr := sched.Cache.ForgetPod(assumedPod); forgetErr != nil {
                klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
            } else {
                // "Forget"ing an assumed Pod in binding cycle should be treated as a PodDelete event,
                // as the assumed Pod had occupied a certain amount of resources in scheduler cache.
                // TODO(#103853): de-duplicate the logic.
                // Avoid moving the assumed Pod itself as it's always Unschedulable.
                // It's intentional to "defer" this operation; otherwise MoveAllToActiveOrBackoffQueue() would
                // update `q.moveRequest` and thus move the assumed pod to backoffQ anyways.
                defer sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, func(pod *v1.Pod) bool {
                    return assumedPod.UID != pod.UID
                })
            }
            sched.handleSchedulingFailure(ctx, fwk, assumedPodInfo, waitOnPermitStatus.AsError(), reason, clearNominatedNode)
            return
        }

    // 运行Prebind 插件
        preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
        if !preBindStatus.IsSuccess() {
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
            // trigger un-reserve plugins to clean up state associated with the reserved Pod
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
            if forgetErr := sched.Cache.ForgetPod(assumedPod); forgetErr != nil {
                klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
            } else {
                // "Forget"ing an assumed Pod in binding cycle should be treated as a PodDelete event,
                // as the assumed Pod had occupied a certain amount of resources in scheduler cache.
                // TODO(#103853): de-duplicate the logic.
                sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, nil)
            }
            sched.handleSchedulingFailure(ctx, fwk, assumedPodInfo, preBindStatus.AsError(), SchedulerError, clearNominatedNode)
            return
        }
        // bind是真正的绑定操作
        err := sched.bind(bindingCycleCtx, fwk, assumedPod, scheduleResult.SuggestedHost, state)
        if err != nil {
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
            // 如果失败了就触发 un-reserve plugins 
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
            if err := sched.Cache.ForgetPod(assumedPod); err != nil {
                klog.ErrorS(err, "scheduler cache ForgetPod failed")
            } else {
                // "Forget"ing an assumed Pod in binding cycle should be treated as a PodDelete event,
                // as the assumed Pod had occupied a certain amount of resources in scheduler cache.
                // TODO(#103853): de-duplicate the logic.
                sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, nil)
            }
            sched.handleSchedulingFailure(ctx, fwk, assumedPodInfo, fmt.Errorf("binding rejected: %w", err), SchedulerError, clearNominatedNode)
            return
        }
        // Calculating nodeResourceString can be heavy. Avoid it if klog verbosity is below 2.
        klog.V(2).InfoS("Successfully bound pod to node", "pod", klog.KObj(pod), "node", scheduleResult.SuggestedHost, "evaluatedNodes", scheduleResult.EvaluatedNodes, "feasibleNodes", scheduleResult.FeasibleNodes)
        metrics.PodScheduled(fwk.ProfileName(), metrics.SinceInSeconds(start))
        metrics.PodSchedulingAttempts.Observe(float64(podInfo.Attempts))
        metrics.PodSchedulingDuration.WithLabelValues(getAttemptsLabel(podInfo)).Observe(metrics.SinceInSeconds(podInfo.InitialAttemptTimestamp))

        // 运行 "postbind" 插件
        // 是通知性的扩展点，该插件在绑定 Pod 后调用，可用于清理相关资源（）。
        fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

        // At the end of a successful binding cycle, move up Pods if needed.
        if len(podsToActivate.Map) != 0 {
            sched.SchedulingQueue.Activate(podsToActivate.Map)
            // Unlike the logic in scheduling cycle, we don't bother deleting the entries
            // as `podsToActivate.Map` is no longer consumed.
        }
    }()
}

调度上下文中的失败流程

上面说到的都是正常的请求，下面会对失败的请求是如何重试的进行分析，而 scheduler 中关于失败处理方面相关的属性会涉及到上面 scheduler 结构中的 backoffQ 与 unschedulablePods

backoffQ：也是一个 heap 类型的优先级队列，存放的是不可调度的Pod
unschedulablePods：保存确定不可被调度的Pod，一个map类型

backoffQ 与 unschedulablePods 会在初始化 scheduler 时初始化，

func NewPriorityQueue(
    lessFn framework.LessFunc,
    informerFactory informers.SharedInformerFactory,
    opts ...Option,
) *PriorityQueue {
    options := defaultPriorityQueueOptions
    for _, opt := range opts {
        opt(&options)
    }

    comp := func(podInfo1, podInfo2 interface{}) bool {
        pInfo1 := podInfo1.(*framework.QueuedPodInfo)
        pInfo2 := podInfo2.(*framework.QueuedPodInfo)
        return lessFn(pInfo1, pInfo2)
    }

    if options.podNominator == nil {
        options.podNominator = NewPodNominator(informerFactory.Core().V1().Pods().Lister())
    }

    pq := &PriorityQueue{
        PodNominator:                      options.podNominator,
        clock:                             options.clock,
        stop:                              make(chan struct{}),
        podInitialBackoffDuration:         options.podInitialBackoffDuration,
        podMaxBackoffDuration:             options.podMaxBackoffDuration,
        podMaxInUnschedulablePodsDuration: options.podMaxInUnschedulablePodsDuration,
        activeQ:                           heap.NewWithRecorder(podInfoKeyFunc, comp, metrics.NewActivePodsRecorder()),
        unschedulablePods:                 newUnschedulablePods(metrics.NewUnschedulablePodsRecorder()),
        moveRequestCycle:                  -1,
        clusterEventMap:                   options.clusterEventMap,
    }
    pq.cond.L = &pq.lock
    // 初始化backoffQ
    // NewWithRecorder作为一个可选的 metricRecorder 的 Heap 对象。
    // podInfoKeyFunc是一个函数，返回错误与字符串
    // pq.podsCompareBackoffCompleted 比较两个pod的回退时间，如果第一个在第二个之前为true，
    // 反之 false
    pq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, pq.podsCompareBackoffCompleted, metrics.NewBackoffPodsRecorder())
    pq.nsLister = informerFactory.Core().V1().Namespaces().Lister()

    return pq
}

对于初始化 backoffQ 会产生的两个函数，getBackoffTime 与 calculateBackoffDuration

// getBackoffTime returns the time that podInfo completes backoff
func (p *PriorityQueue) getBackoffTime(podInfo *framework.QueuedPodInfo) time.Time {
    duration := p.calculateBackoffDuration(podInfo)
    backoffTime := podInfo.Timestamp.Add(duration)
    return backoffTime
}

// calculateBackoffDuration is a helper function for calculating the backoffDuration
// based on the number of attempts the pod has made.
func (p *PriorityQueue) calculateBackoffDuration(podInfo *framework.QueuedPodInfo) time.Duration {
    duration := p.podInitialBackoffDuration
    for i := 1; i < podInfo.Attempts; i++ {
        // Use subtraction instead of addition or multiplication to avoid overflow.
        if duration > p.podMaxBackoffDuration-duration {
            return p.podMaxBackoffDuration
        }
        duration += duration
    }
    return duration
}

对于整个故障错误会按照如下流程进行，在初始化 scheduler 会注册一个 Error 函数，这个函数用作对不可调度Pod进行处理，实际上被注册的函数是 MakeDefaultErrorFunc。这个函数将作为 Error 函数被调用。

sched := newScheduler(
    schedulerCache,
    extenders,
    internalqueue.MakeNextPodFunc(podQueue),
    MakeDefaultErrorFunc(client, podLister, podQueue, schedulerCache),
    stopEverything,
    podQueue,
    profiles,
    client,
    snapshot,
    options.percentageOfNodesToScore,
)

而在调度周期中，也就是 scheduleOne 可以看到，每个扩展点操作失败后都会调用 handleSchedulingFailure 而该函数，使用了注册的 Error 函数来处理Pod

func (sched *Scheduler) scheduleOne(ctx context.Context) {
    ...
    defer cancel()
    scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)
    if err != nil {

        var nominatingInfo *framework.NominatingInfo
        if fitError, ok := err.(*framework.FitError); ok {
            if !fwk.HasPostFilterPlugins() {
                klog.V(3).InfoS("No PostFilter plugins are registered, so no preemption will be performed")
            } else {

                result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap)
                if status.Code() == framework.Error {
                    klog.ErrorS(nil, "Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                } else {
                    fitError.Diagnosis.PostFilterMsg = status.Message()
                    klog.V(5).InfoS("Status after running PostFilter plugins for pod", "pod", klog.KObj(pod), "status", status)
                }
                if result != nil {
                    nominatingInfo = result.NominatingInfo
                }
            }

            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else if err == ErrNoNodesAvailable {
            nominatingInfo = clearNominatedNode
            // No nodes available is counted as unschedulable rather than an error.
            metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        } else {
            nominatingInfo = clearNominatedNode
            klog.ErrorS(err, "Error selecting node for pod", "pod", klog.KObj(pod))
            metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        }
        // 处理不可调度Pod
        sched.handleSchedulingFailure(ctx, fwk, podInfo, err, v1.PodReasonUnschedulable, nominatingInfo)
        return
    }

来到了注册的 Error 函数 MakeDefaultErrorFunc

func MakeDefaultErrorFunc(client clientset.Interface, podLister corelisters.PodLister, podQueue internalqueue.SchedulingQueue, schedulerCache internalcache.Cache) func(*framework.QueuedPodInfo, error) {
    return func(podInfo *framework.QueuedPodInfo, err error) {
        pod := podInfo.Pod
        if err == ErrNoNodesAvailable {
            klog.V(2).InfoS("Unable to schedule pod; no nodes are registered to the cluster; waiting", "pod", klog.KObj(pod))
        } else if fitError, ok := err.(*framework.FitError); ok {
            // Inject UnschedulablePlugins to PodInfo, which will be used later for moving Pods between queues efficiently.
            podInfo.UnschedulablePlugins = fitError.Diagnosis.UnschedulablePlugins
            klog.V(2).InfoS("Unable to schedule pod; no fit; waiting", "pod", klog.KObj(pod), "err", err)
        } else if apierrors.IsNotFound(err) {
            klog.V(2).InfoS("Unable to schedule pod, possibly due to node not found; waiting", "pod", klog.KObj(pod), "err", err)
            if errStatus, ok := err.(apierrors.APIStatus); ok && errStatus.Status().Details.Kind == "node" {
                nodeName := errStatus.Status().Details.Name
                // when node is not found, We do not remove the node right away. Trying again to get
                // the node and if the node is still not found, then remove it from the scheduler cache.
                _, err := client.CoreV1().Nodes().Get(context.TODO(), nodeName, metav1.GetOptions{})
                if err != nil && apierrors.IsNotFound(err) {
                    node := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}
                    if err := schedulerCache.RemoveNode(&node); err != nil {
                        klog.V(4).InfoS("Node is not found; failed to remove it from the cache", "node", node.Name)
                    }
                }
            }
        } else {
            klog.ErrorS(err, "Error scheduling pod; retrying", "pod", klog.KObj(pod))
        }

        // Check if the Pod exists in informer cache.
        cachedPod, err := podLister.Pods(pod.Namespace).Get(pod.Name)
        if err != nil {
            klog.InfoS("Pod doesn't exist in informer cache", "pod", klog.KObj(pod), "err", err)
            return
        }

        // In the case of extender, the pod may have been bound successfully, but timed out returning its response to the scheduler.
        // It could result in the live version to carry .spec.nodeName, and that's inconsistent with the internal-queued version.
        if len(cachedPod.Spec.NodeName) != 0 {
            klog.InfoS("Pod has been assigned to node. Abort adding it back to queue.", "pod", klog.KObj(pod), "node", cachedPod.Spec.NodeName)
            return
        }

        // As <cachedPod> is from SharedInformer, we need to do a DeepCopy() here.
        podInfo.PodInfo = framework.NewPodInfo(cachedPod.DeepCopy())
        // 添加到unschedulable队列中
        if err := podQueue.AddUnschedulableIfNotPresent(podInfo, podQueue.SchedulingCycle()); err != nil {
            klog.ErrorS(err, "Error occurred")
        }
    }
}

下面来到 AddUnschedulableIfNotPresent ，这个也是操作 backoffQ 和 unschedulablePods 的真正的动作

AddUnschedulableIfNotPresent 函数会吧无法调度的 pod 插入队列，除非它已经在队列中。通常情况下，PriorityQueue 将不可调度的 Pod 放在 unschedulablePods 中。但如果最近有 move request，则将 pod 放入 podBackoffQ 中。

func (p *PriorityQueue) AddUnschedulableIfNotPresent(pInfo *framework.QueuedPodInfo, podSchedulingCycle int64) error {
    p.lock.Lock()
    defer p.lock.Unlock()
    pod := pInfo.Pod
    // 如果已经存在则不添加
    if p.unschedulablePods.get(pod) != nil {
        return fmt.Errorf("Pod %v is already present in unschedulable queue", klog.KObj(pod))
    }
    // 检查是否在activeQ中
    if _, exists, _ := p.activeQ.Get(pInfo); exists {
        return fmt.Errorf("Pod %v is already present in the active queue", klog.KObj(pod))
    }
    // 检查是否在podBackoffQ中
    if _, exists, _ := p.podBackoffQ.Get(pInfo); exists {
        return fmt.Errorf("Pod %v is already present in the backoff queue", klog.KObj(pod))
    }

    // 在重新添加时，会刷新 Pod时间为最新操作的时间
    pInfo.Timestamp = p.clock.Now()

    for plugin := range pInfo.UnschedulablePlugins {
        metrics.UnschedulableReason(plugin, pInfo.Pod.Spec.SchedulerName).Inc()
    }
    // 如果接受到move request那么则放入BackoffQ
    if p.moveRequestCycle >= podSchedulingCycle {
        if err := p.podBackoffQ.Add(pInfo); err != nil {
            return fmt.Errorf("error adding pod %v to the backoff queue: %v", pod.Name, err)
        }
        metrics.SchedulerQueueIncomingPods.WithLabelValues("backoff", ScheduleAttemptFailure).Inc()
    } else {
        // 否则将放入到 unschedulablePods
        p.unschedulablePods.addOrUpdate(pInfo)
        metrics.SchedulerQueueIncomingPods.WithLabelValues("unschedulable", ScheduleAttemptFailure).Inc()

    }

    p.PodNominator.AddNominatedPod(pInfo.PodInfo, nil)
    return nil
}

在启动 scheduler 时，会将这两个队列异步启用两个loop来操作队列。表现在 Run()

func (p *PriorityQueue) Run() {
    go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)
    go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop)
}

可以看到 flushBackoffQCompleted 作为 BackoffQ 实现；而 flushUnschedulablePodsLeftover 作为 UnschedulablePods 实现。

flushBackoffQCompleted 是用于将所有已完成回退的 pod 从 backoffQ 移到 activeQ 中

func (p *PriorityQueue) flushBackoffQCompleted() {
    p.lock.Lock()
    defer p.lock.Unlock()
    broadcast := false
    for { // 这就是heap实现的方法，窥视下，但不弹出
        rawPodInfo := p.podBackoffQ.Peek()
        if rawPodInfo == nil {
            break
        }
        pod := rawPodInfo.(*framework.QueuedPodInfo).Pod
        boTime := p.getBackoffTime(rawPodInfo.(*framework.QueuedPodInfo))
        if boTime.After(p.clock.Now()) {
            break
        }
        _, err := p.podBackoffQ.Pop() // 弹出一个
        if err != nil {
            klog.ErrorS(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod))
            break
        }
        p.activeQ.Add(rawPodInfo) // 放入到活动队列中
        metrics.SchedulerQueueIncomingPods.WithLabelValues("active", BackoffComplete).Inc()
        broadcast = true
    }

    if broadcast {
        p.cond.Broadcast()
    }
}

flushUnschedulablePodsLeftover 函数用于将在 unschedulablePods 中的存放时间超过 podMaxInUnschedulablePodsDuration 值的 pod 移动到 backoffQ 或 activeQ 中。

podMaxInUnschedulablePodsDuration 会根据配置传入，当没有传入，也就是使用了 Deprecated 那么会为5分钟。

func NewOptions() *Options {
    o := &Options{
        SecureServing:  apiserveroptions.NewSecureServingOptions().WithLoopback(),
        Authentication: apiserveroptions.NewDelegatingAuthenticationOptions(),
        Authorization:  apiserveroptions.NewDelegatingAuthorizationOptions(),
        Deprecated: &DeprecatedOptions{
            PodMaxInUnschedulablePodsDuration: 5 * time.Minute,
        },

对于 flushUnschedulablePodsLeftover 就是做一个时间对比，然后添加到对应的队列中

func (p *PriorityQueue) flushUnschedulablePodsLeftover() {
    p.lock.Lock()
    defer p.lock.Unlock()

    var podsToMove []*framework.QueuedPodInfo
    currentTime := p.clock.Now()
    for _, pInfo := range p.unschedulablePods.podInfoMap {
        lastScheduleTime := pInfo.Timestamp
        if currentTime.Sub(lastScheduleTime) > p.podMaxInUnschedulablePodsDuration {
            podsToMove = append(podsToMove, pInfo)
        }
    }

    if len(podsToMove) > 0 {
        p.movePodsToActiveOrBackoffQueue(podsToMove, UnschedulableTimeout)
    }
}

总结调度上下文流程

在构建一个 scheduler 时经历如下步骤：
- 准备cache，informer，queue，错误处理函数等
- 添加事件函数，会监听资源（如Pod），当有变动则触发对应事件函数，这是入站 activeQ
构建完成后会 run，run时会run一个 SchedulingQueue，这个是作为不可调度队列
- BackoffQ
- UnschedulablePods
- 不可调度队列会根据注册时定期消费队列中Pod将其添加到 activeQ 中
启动一个 scheduleOne 的loop，这个是调度上下文中所有的扩展点的执行，也是 activeQ 的消费端
- scheduleOne 获取 pod
- 执行各个扩展点，如果出错则 Error 函数 MakeDefaultErrorFunc 将其添加到不可调度队列中
- 回到不可调度队列中消费部分

Reference

^[1] kubernetes scheduler extender
^[2] scheduling framework
^[3] Extension points

彻底搞懂kubernetes调度框架与插件

cylon — Sun, 31 Jul 2022 14:33:53 +0000

调度框架 ^[1]

本文基于 kubernetes 1.24 进行分析

调度框架（Scheduling Framework）是Kubernetes 的调度器 kube-scheduler 设计的的可插拔架构，将插件（调度算法）嵌入到调度上下文的每个扩展点中，并编译为 kube-scheduler

在 kube-scheduler 1.22 之后，在 pkg/scheduler/framework/interface.go 中定义了一个 Plugin 的 interface，这个 interface 作为了所有插件的父级。而每个未调度的 Pod，Kubernetes 调度器会根据一组规则尝试在集群中寻找一个节点。

type Plugin interface {
    Name() string
}

下面会对每个算法是如何实现的进行分析

在初始化 scheduler 时，会创建一个 profile，profile是关于 scheduler 调度配置相关的定义

func New(client clientset.Interface,
...
    profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
        frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
        frameworkruntime.WithClientSet(client),
        frameworkruntime.WithKubeConfig(options.kubeConfig),
        frameworkruntime.WithInformerFactory(informerFactory),
        frameworkruntime.WithSnapshotSharedLister(snapshot),
        frameworkruntime.WithPodNominator(nominator),
        frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
        frameworkruntime.WithClusterEventMap(clusterEventMap),
        frameworkruntime.WithParallelism(int(options.parallelism)),
        frameworkruntime.WithExtenders(extenders),
    )
    if err != nil {
        return nil, fmt.Errorf("initializing profiles: %v", err)
    }

    if len(profiles) == 0 {
        return nil, errors.New("at least one profile is required")
    }
....
}

关于 profile 的实现，则为 KubeSchedulerProfile，也是作为 yaml生成时传入的配置

// KubeSchedulerProfile 是一个 scheduling profile.
type KubeSchedulerProfile struct {
    // SchedulerName 是与此配置文件关联的调度程序的名称。
    // 如果 SchedulerName 与 pod “spec.schedulerName”匹配，则使用此配置文件调度 pod。
    SchedulerName string

    // Plugins指定应该启用或禁用的插件集。
    // 启用的插件是除了默认插件之外应该启用的插件。禁用插件应是禁用的任何默认插件。
    // 当没有为扩展点指定启用或禁用插件时，将使用该扩展点的默认插件（如果有）。
    // 如果指定了 QueueSort 插件，
    // 则必须为所有配置文件指定相同的 QueueSort Plugin 和 PluginConfig。
    // 这个Plugins展现的形式则是调度上下文中的所有扩展点(这是抽象)，实际中会表现为多个扩展点
    Plugins *Plugins

    // PluginConfig 是每个插件的一组可选的自定义插件参数。
    // 如果省略PluginConfig参数等同于使用该插件的默认配置。
    PluginConfig []PluginConfig
}

对于 profile.NewMap 就是根据给定的配置来构建这个framework，因为配置可能是存在多个的。而 Registry 则是所有可用插件的集合，内部构造则是 PluginFactory ,通过函数来构建出对应的 plugin

func NewMap(cfgs []config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
    stopCh <-chan struct{}, opts ...frameworkruntime.Option) (Map, error) {
    m := make(Map)
    v := cfgValidator{m: m}

    for _, cfg := range cfgs {
        p, err := newProfile(cfg, r, recorderFact, stopCh, opts...)
        if err != nil {
            return nil, fmt.Errorf("creating profile for scheduler name %s: %v", cfg.SchedulerName, err)
        }
        if err := v.validate(cfg, p); err != nil {
            return nil, err
        }
        m[cfg.SchedulerName] = p
    }
    return m, nil
}

// newProfile 给的配置构建出一个profile
func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
    stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
    recorder := recorderFact(cfg.SchedulerName)
    opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
    return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}

可以看到最终返回的是一个 Framework 。那么来看下这个 Framework

Framework 是一个抽象，管理着调度过程中所使用的所有插件，并在调度上下文中适当的位置去运行对应的插件

type Framework interface {
    Handle
    // QueueSortFunc 返回对调度队列中的 Pod 进行排序的函数
    // 也就是less，在Sort打分阶段的打分函数
    QueueSortFunc() LessFunc

    // RunPreFilterPlugins 运行配置的一组PreFilter插件。
    // 如果这组插件中，任何一个插件失败，则返回 *Status 并设置为non-success。
    // 如果返回状态为non-success，则调度周期中止。
    // 它还返回一个 PreFilterResult，它可能会影响到要评估下游的节点。

    RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)

    // RunPostFilterPlugins 运行配置的一组PostFilter插件。 
    // PostFilter 插件是通知性插件，在这种情况下应配置为先执行并返回 Unschedulable 状态，
    // 或者尝试更改集群状态以使 pod 在未来的调度周期中可能会被调度。
    RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)

    // RunPreBindPlugins 运行配置的一组 PreBind 插件。
    // 如果任何一个插件返回错误，则返回 *Status 并且code设置为non-success。
    // 如果code为“Unschedulable”，则调度检查失败，
    // 则认为是内部错误。在任何一种情况下，Pod都不会被bound。
    RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // RunPostBindPlugins 运行配置的一组PostBind插件
    RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)

    // RunReservePluginsReserve运行配置的一组Reserve插件的Reserve方法。
    // 如果在这组调用中的任何一个插件返回错误，则不会继续运行剩余调用的插件并返回错误。
    // 在这种情况下，pod将不能被调度。
    RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // RunReservePluginsUnreserve运行配置的一组Reserve插件的Unreserve方法。
    RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)

    // RunPermitPlugins运行配置的一组Permit插件。
    // 如果这些插件中的任何一个返回“Success”或“Wait”之外的状态，则它不会继续运行其余插件并返回错误。
    // 否则，如果任何插件返回 “Wait”，则此函数将创建等待pod并将其添加到当前等待pod的map中，
    // 并使用“Wait” code返回状态。 Pod将在Permit插件返回的最短持续时间内保持等待pod。
    RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // 如果pod是waiting pod，WaitOnPermit 将阻塞，直到等待的pod被允许或拒绝。
    WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status

    // RunBindPlugins运行配置的一组bind插件。 Bind插件可以选择是否处理Pod。
    // 如果 Bind 插件选择跳过binding，它应该返回 code=5("skip")状态。
    // 否则，它应该返回“Error”或“Success”。
    // 如果没有插件处理绑定，则RunBindPlugins返回code=5("skip")的状态。
    RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status

    // 如果至少定义了一个filter插件，则HasFilterPlugins返回true
    HasFilterPlugins() bool

    // 如果至少定义了一个PostFilter插件，则HasPostFilterPlugins返回 true。
    HasPostFilterPlugins() bool

    // 如果至少定义了一个Score插件，则HasScorePlugins返回 true。
    HasScorePlugins() bool

    // ListPlugins将返回map。key为扩展点名称，value则是配置的插件列表。
    ListPlugins() *config.Plugins

    // ProfileName则是与profile name关联的framework
    ProfileName() string
}

而实现这个抽象的则是 frameworkImpl；frameworkImpl 是初始化与运行 scheduler plugins 的组件，并在调度上下文中会运行这些扩展点

type frameworkImpl struct {
   registry             Registry
   snapshotSharedLister framework.SharedLister
   waitingPods          *waitingPodsMap
   scorePluginWeight    map[string]int
   queueSortPlugins     []framework.QueueSortPlugin
   preFilterPlugins     []framework.PreFilterPlugin
   filterPlugins        []framework.FilterPlugin
   postFilterPlugins    []framework.PostFilterPlugin
   preScorePlugins      []framework.PreScorePlugin
   scorePlugins         []framework.ScorePlugin
   reservePlugins       []framework.ReservePlugin
   preBindPlugins       []framework.PreBindPlugin
   bindPlugins          []framework.BindPlugin
   postBindPlugins      []framework.PostBindPlugin
   permitPlugins        []framework.PermitPlugin

   clientSet       clientset.Interface
   kubeConfig      *restclient.Config
   eventRecorder   events.EventRecorder
   informerFactory informers.SharedInformerFactory

   metricsRecorder *metricsRecorder
   profileName     string

   extenders []framework.Extender
   framework.PodNominator

   parallelizer parallelize.Parallelizer
}

那么来看下 Registry ，Registry 是作为一个可用插件的集合。framework 使用 registry 来启用和对插件配置的初始化。在初始化框架之前，所有插件都必须在注册表中。表现形式就是一个 map[]；key 是插件的名称，value是 PluginFactory 。

type Registry map[string]PluginFactory

而在 pkg\scheduler\framework\plugins\registry.go 中会将所有的 in-tree plugin 注册进来。通过 NewInTreeRegistry 。后续如果还有插件要注册，可以通过 WithFrameworkOutOfTreeRegistry 来注册其他的插件。

func NewInTreeRegistry() runtime.Registry {
    fts := plfeature.Features{
        EnableReadWriteOncePod:                       feature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
        EnableVolumeCapacityPriority:                 feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),
        EnableMinDomainsInPodTopologySpread:          feature.DefaultFeatureGate.Enabled(features.MinDomainsInPodTopologySpread),
        EnableNodeInclusionPolicyInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.NodeInclusionPolicyInPodTopologySpread),
    }

    return runtime.Registry{
        selectorspread.Name:                  selectorspread.New,
        imagelocality.Name:                   imagelocality.New,
        tainttoleration.Name:                 tainttoleration.New,
        nodename.Name:                        nodename.New,
        nodeports.Name:                       nodeports.New,
        nodeaffinity.Name:                    nodeaffinity.New,
        podtopologyspread.Name:               runtime.FactoryAdapter(fts, podtopologyspread.New),
        nodeunschedulable.Name:               nodeunschedulable.New,
        noderesources.Name:                   runtime.FactoryAdapter(fts, noderesources.NewFit),
        noderesources.BalancedAllocationName: runtime.FactoryAdapter(fts, noderesources.NewBalancedAllocation),
        volumebinding.Name:                   runtime.FactoryAdapter(fts, volumebinding.New),
        volumerestrictions.Name:              runtime.FactoryAdapter(fts, volumerestrictions.New),
        volumezone.Name:                      volumezone.New,
        nodevolumelimits.CSIName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewCSI),
        nodevolumelimits.EBSName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewEBS),
        nodevolumelimits.GCEPDName:           runtime.FactoryAdapter(fts, nodevolumelimits.NewGCEPD),
        nodevolumelimits.AzureDiskName:       runtime.FactoryAdapter(fts, nodevolumelimits.NewAzureDisk),
        nodevolumelimits.CinderName:          runtime.FactoryAdapter(fts, nodevolumelimits.NewCinder),
        interpodaffinity.Name:                interpodaffinity.New,
        queuesort.Name:                       queuesort.New,
        defaultbinder.Name:                   defaultbinder.New,
        defaultpreemption.Name:               runtime.FactoryAdapter(fts, defaultpreemption.New),
    }
}

这里插入一个题外话，关于 in-tree plugin

在这里没有找到关于，kube-scheduler ，只是找到有关的概念，大概可以解释为，in-tree表示为随kubernetes官方提供的二进制构建的 plugin 则为 in-tree，而独立于kubernetes代码库之外的为 out-of-tree ^[3] 。这种情况下，可以理解为，AA则是 out-of-tree 而 Pod, DeplymentSet 等是 in-tree。

接下来回到初始化 scheduler ，在初始化一个 scheduler 时，会通过NewInTreeRegistry 来初始化

func New(client clientset.Interface,
    ....
    registry := frameworkplugins.NewInTreeRegistry()
    if err := registry.Merge(options.frameworkOutOfTreeRegistry); err != nil {
        return nil, err
    }

    ...

    profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
        frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
        frameworkruntime.WithClientSet(client),
        frameworkruntime.WithKubeConfig(options.kubeConfig),
        frameworkruntime.WithInformerFactory(informerFactory),
        frameworkruntime.WithSnapshotSharedLister(snapshot),
        frameworkruntime.WithPodNominator(nominator),
        frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
        frameworkruntime.WithClusterEventMap(clusterEventMap),
        frameworkruntime.WithParallelism(int(options.parallelism)),
        frameworkruntime.WithExtenders(extenders),
    )
    ...
}

接下来在调度上下文 scheduleOne 中 schedulePod 时，会通过 framework 调用对应的插件来处理这个扩展点工作。具体的体现在，pkg\scheduler\schedule_one.go 中的预选阶段

func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
    trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
    defer trace.LogIfLong(100 * time.Millisecond)

    if err := sched.Cache.UpdateSnapshot(sched.nodeInfoSnapshot); err != nil {
        return result, err
    }
    trace.Step("Snapshotting scheduler cache and node infos done")

    if sched.nodeInfoSnapshot.NumNodes() == 0 {
        return result, ErrNoNodesAvailable
    }

    feasibleNodes, diagnosis, err := sched.findNodesThatFitPod(ctx, fwk, state, pod)
    if err != nil {
        return result, err
    }
    trace.Step("Computing predicates done")

与其他扩展点部分，在调度上下文 scheduleOne 中可以很好的看出，功能都是 framework 提供的。

func (sched *Scheduler) scheduleOne(ctx context.Context) {

    ...

    scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)

    ...

    // Run the Reserve method of reserve plugins.
    if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
    }

    ...

    // Run "permit" plugins.
    runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

        // One of the plugins returned status different than success or wait.
        fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

...

    // bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
    go func() {
        ...
        waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
        if !waitOnPermitStatus.IsSuccess() {
            ...
            // trigger un-reserve plugins to clean up state associated with the reserved Pod
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
        }

        // Run "prebind" plugins.
        preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

        ...

            // trigger un-reserve plugins to clean up state associated with the reserved Pod
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

        ...

        ...
            // trigger un-reserve plugins to clean up state associated with the reserved Pod
            fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

        ...

        // Run "postbind" plugins.
        fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

    ...
}

插件 ^[4]

插件（Plugins）（也可以算是调度策略）在 kube-scheduler 中的实现为 framework plugin，插件API的实现分为两个步骤：register 和 configured，然后都实现了其父方法 Plugin。然后可以通过配置（kube-scheduler --config 提供）启动或禁用插件；除了默认插件外，还可以实现自定义调度插件与默认插件进行绑定。

type Plugin interface {
    Name() string
}
// sort扩展点
type QueueSortPlugin interface {
    Plugin
    Less(*v1.pod, *v1.pod) bool
}
// PreFilter扩展点
type PreFilterPlugin interface {
    Plugin
    PreFilter(context.Context, *framework.CycleState, *v1.pod) error
}

插件的载入过程

在 scheduler 被启动时，会 scheduler.New(cc.Client.. 这个时候会传入 profiles，整个的流如下：

NewScheduler ：kubernetes/cmd/kube-scheduler/app/server.go
profile.NewMap：kubernetes/pkg/scheduler/scheduler.go
- newProfile：kubernetes/pkg/scheduler/scheduler.go
frameworkruntime.NewFramework：kubernetes/pkg/scheduler/framework/runtime/framework.go
- pluginsNeeded：kubernetes/pkg/scheduler/framework/runtime/framework.go

NewScheduler

我们了解如何 New 一个 scheduler 即为 Setup 中去配置这些参数，

func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions ...Option) (*schedulerserverconfig.CompletedConfig, *scheduler.Scheduler, error) {

    ...

    // Create the scheduler.
    sched, err := scheduler.New(cc.Client,
        cc.InformerFactory,
        cc.DynInformerFactory,
        recorderFactory,
        ctx.Done(),
        scheduler.WithComponentConfigVersion(cc.ComponentConfig.TypeMeta.APIVersion),
        scheduler.WithKubeConfig(cc.KubeConfig),
        scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
        scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
        scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
        scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
        scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
        scheduler.WithPodMaxInUnschedulablePodsDuration(cc.PodMaxInUnschedulablePodsDuration),
        scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
        scheduler.WithParallelism(cc.ComponentConfig.Parallelism),
        scheduler.WithBuildFrameworkCapturer(func(profile kubeschedulerconfig.KubeSchedulerProfile) {
            // Profiles are processed during Framework instantiation to set default plugins and configurations. Capturing them for logging
            completedProfiles = append(completedProfiles, profile)
        }),
    )
    ...
}

profile.NewMap

在 scheduler.New 中，会根据配置生成profile，而 profile.NewMap 会完成这一步

func New(client clientset.Interface,
    ...

    clusterEventMap := make(map[framework.ClusterEvent]sets.String)

    profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
        frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
        frameworkruntime.WithClientSet(client),
        frameworkruntime.WithKubeConfig(options.kubeConfig),
        frameworkruntime.WithInformerFactory(informerFactory),
        frameworkruntime.WithSnapshotSharedLister(snapshot),
        frameworkruntime.WithPodNominator(nominator),
        frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
        frameworkruntime.WithClusterEventMap(clusterEventMap),
        frameworkruntime.WithParallelism(int(options.parallelism)),
        frameworkruntime.WithExtenders(extenders),
    )

         ...
}

NewFramework

newProfile 返回的则是一个创建好的 framework

func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
    stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
    recorder := recorderFact(cfg.SchedulerName)
    opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
    return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}

最终会走到 pluginsNeeded，这里会根据配置中开启的插件而返回一个插件集，这个就是最终在每个扩展点中要执行的插件。

func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) sets.String {
    pgSet := sets.String{}

    if plugins == nil {
        return pgSet
    }

    find := func(pgs *config.PluginSet) {
        for _, pg := range pgs.Enabled {
            pgSet.Insert(pg.Name)
        }
    }
    // 获取到所有的扩展点，找到为Enabled的插件加入到pgSet
    for _, e := range f.getExtensionPoints(plugins) {
        find(e.plugins)
    }
    // Parse MultiPoint separately since they are not returned by f.getExtensionPoints()
    find(&plugins.MultiPoint)

    return pgSet
}

插件的执行

在对插件源码部分分析，会找几个典型的插件进行分析，而不会对全部的进行分析，因为总的来说是大同小异，分析的插件有 NodePorts，NodeResourcesFit，podtopologyspread

NodePorts

这里以一个简单的插件来分析；NodePorts 插件用于检查Pod请求的端口，在节点上是否为空闲端口。

NodePorts 实现了 FilterPlugin 和 PreFilterPlugin

PreFilter 将会被 framework 中 PreFilter 扩展点被调用。

func (pl *NodePorts) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
    s := getContainerPorts(pod) // 或得Pod得端口
    // 写入状态
    cycleState.Write(preFilterStateKey, preFilterState(s))
    return nil, nil
}

Filter 将会被 framework 中 Filter 扩展点被调用。

// Filter invoked at the filter extension point.
func (pl *NodePorts) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
   wantPorts, err := getPreFilterState(cycleState)
   if err != nil {
      return framework.AsStatus(err)
   }

   fits := fitsPorts(wantPorts, nodeInfo)
   if !fits {
      return framework.NewStatus(framework.Unschedulable, ErrReason)
   }

   return nil
}

func fitsPorts(wantPorts []*v1.ContainerPort, nodeInfo *framework.NodeInfo) bool {
    // 对比existingPorts 和 wantPorts是否冲突，冲突则调度失败
    existingPorts := nodeInfo.UsedPorts
    for _, cp := range wantPorts {
        if existingPorts.CheckConflict(cp.HostIP, string(cp.Protocol), cp.HostPort) {
            return false
        }
    }
    return true
}

New ，初始化新插件，在 register 中注册得

func New(_ runtime.Object, _ framework.Handle) (framework.Plugin, error) {
    return &NodePorts{}, nil
}

在调用中，如果有任何一个插件返回错误，则跳过该扩展点注册得其他插件，返回失败。

func (f *frameworkImpl) RunFilterPlugins(
    ctx context.Context,
    state *framework.CycleState,
    pod *v1.Pod,
    nodeInfo *framework.NodeInfo,
) framework.PluginToStatus {
    statuses := make(framework.PluginToStatus)
    for _, pl := range f.filterPlugins {
        pluginStatus := f.runFilterPlugin(ctx, pl, state, pod, nodeInfo)
        if !pluginStatus.IsSuccess() {
            if !pluginStatus.IsUnschedulable() 
                errStatus := framework.AsStatus(fmt.Errorf("running %q filter plugin: %w", pl.Name(), pluginStatus.AsError())).WithFailedPlugin(pl.Name())
                return map[string]*framework.Status{pl.Name(): errStatus}
            }
            pluginStatus.SetFailedPlugin(pl.Name())
            statuses[pl.Name()] = pluginStatus
        }
    }

    return statuses
}

返回得状态是一个 Status 结构体，该结构体表示了插件运行的结果。由 Code、reasons、（可选）err 和 failedPlugin （失败的那个插件名）组成。当 code 不是 Success 时，应说明原因。而且，当 code 为 Success 时，其他所有字段都应为空。nil 状态也被视为成功。

type Status struct {
    code    Code
    reasons []string
    err     error
    // failedPlugin is an optional field that records the plugin name a Pod failed by.
    // It's set by the framework when code is Error, Unschedulable or UnschedulableAndUnresolvable.
    failedPlugin string
}

NodeResourcesFit ^[5]

NodeResourcesFit 扩展检查节点是否拥有 Pod 请求的所有资源。分数可以使用以下三种策略之一，扩展点为：preFilter， filter，score

LeastAllocated （默认）
MostAllocated
RequestedToCapacityRatio

Fit

NodeResourcesFit PreFilter 可以看到调用得 computePodResourceRequest

// PreFilter invoked at the prefilter extension point.
func (f *Fit) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
   cycleState.Write(preFilterStateKey, computePodResourceRequest(pod))
   return nil, nil
}

computePodResourceRequest 这里有一个注释，总体解释起来是这样得：computePodResourceRequest ，返回值（ framework.Resource）覆盖了每一个维度中资源的最大宽度。因为将按照 init-containers , containers 得顺序运行，会通过迭代方式收集每个维度中的最大值。计算时会对常规容器的资源向量求和，因为containers 运行会同时运行多个容器。计算示例为：

Pod:
  InitContainers
    IC1:
      CPU: 2
      Memory: 1G
    IC2:
      CPU: 2
      Memory: 3G
  Containers
    C1:
      CPU: 2
      Memory: 1G
    C2:
      CPU: 1
      Memory: 1G

在维度1中（InitContainers）所需资源最大值时，CPU=2, Memory=3G；而维度2（Containers）所需资源最大值为：CPU=2, Memory=1G；那么最终结果为 CPU=3, Memory=3G，因为在维度1，最大资源时Memory=3G；而维度2最大资源是CPU=1+2, Memory=1+1，取每个维度中最大资源最大宽度即为 CPU=3, Memory=3G。

下面则看下代码得实现

func computePodResourceRequest(pod *v1.Pod) *preFilterState {
    result := &preFilterState{}
    for _, container := range pod.Spec.Containers {
        result.Add(container.Resources.Requests)
    }

    // 取最大得资源
    for _, container := range pod.Spec.InitContainers {
        result.SetMaxResource(container.Resources.Requests)
    }

    // 如果Overhead正在使用，需要将其计算到总资源中
    if pod.Spec.Overhead != nil {
        result.Add(pod.Spec.Overhead)
    }
    return result
}

// SetMaxResource 是比较ResourceList并为每个资源取最大值。
func (r *Resource) SetMaxResource(rl v1.ResourceList) {
    if r == nil {
        return
    }

    for rName, rQuantity := range rl {
        switch rName {
        case v1.ResourceMemory:
            r.Memory = max(r.Memory, rQuantity.Value())
        case v1.ResourceCPU:
            r.MilliCPU = max(r.MilliCPU, rQuantity.MilliValue())
        case v1.ResourceEphemeralStorage:
            if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
                r.EphemeralStorage = max(r.EphemeralStorage, rQuantity.Value())
            }
        default:
            if schedutil.IsScalarResourceName(rName) {
                r.SetScalar(rName, max(r.ScalarResources[rName], rQuantity.Value()))
            }
        }
    }
}

leastAllocate

LeastAllocated 是 NodeResourcesFit 的打分策略，LeastAllocated 打分的标准是更偏向于请求资源较少的Node。将会先计算出Node上调度的pod请求的内存、CPU与其他资源的百分比，然后并根据请求的比例与容量的平均值的最小值进行优先级排序。

计算公式是这样的：$\frac{\frac{cpu((capacity-requested) \times MaxNodeScore \times cpuWeight)}{capacity} + \frac{memory((capacity-requested) \times MaxNodeScore \times memoryWeight}{capacity}) + ...}{weightSum}$

下面来看下实现

func leastResourceScorer(resToWeightMap resourceToWeightMap) func(resourceToValueMap, resourceToValueMap) int64 {
    return func(requested, allocable resourceToValueMap) int64 {
        var nodeScore, weightSum int64
        for resource := range requested {
            weight := resToWeightMap[resource]
            //  计算出的资源分数乘weight
            resourceScore := leastRequestedScore(requested[resource], allocable[resource])
            nodeScore += resourceScore * weight
            weightSum += weight
        }
        if weightSum == 0 {
            return 0
        }
        // 最终除weightSum
        return nodeScore / weightSum
    }
}

leastRequestedScore 计算标准为未使用容量的计算范围为 0~MaxNodeScore，0 为最低优先级，MaxNodeScore 为最高优先级。未使用的资源越多，得分越高。

func leastRequestedScore(requested, capacity int64) int64 {
    if capacity == 0 {
        return 0
    }
    if requested > capacity {
        return 0
    }
    // 容量 - 请求的 x 预期值（100）/ 容量
    return ((capacity - requested) * int64(framework.MaxNodeScore)) / capacity
}

Topology ^[6]

Concept

在对 podtopologyspread 插件进行分析前，先需要掌握Pod拓扑的概念。

Pod拓扑（Pod Topology）是Kubernetes Pod调度机制，可以将Pod分布在集群中不同 Zone ，以及用户自定义的各种拓扑域（topology domains）。当有了拓扑域后，用户可以更高效的利用集群资源。

如何来解释拓扑域，首先需要提及为什么需要拓扑域，在集群有3个节点，并且当Pod副本数为2时，又不希望两个Pod在同一个Node上运行。在随着扩大Pod的规模，副本数扩展到到15个时，这时候最理想的方式是每个Node运行5个Pod，在这种背景下，用户希望对集群中Zone的安排为相似的副本数量，并且在集群存在部分问题时可以更好的自愈（也是按照相似的副本数量均匀的分布在Node上）。在这种情况下Kubernetes 提供了Pod 拓扑约束来解决这个问题。

定义一个Topology

apiVersion: v1
kind: Pod
metadata:
  name: example-pod
spec:
  # Configure a topology spread constraint
  topologySpreadConstraints:
    - maxSkew: <integer> # 
      minDomains: <integer> # optional; alpha since v1.24
      topologyKey: <string>
      whenUnsatisfiable: <string>
      labelSelector: <object>

参数的描述：

maxSkew：Required，Pod分布不均的程度，并且数字必须大于零
- 当 whenUnsatisfiable: DoNotSchedule，则定义目标拓扑中匹配 pod 的数量与 全局最小值（拓扑域中的标签选择器匹配的 pod 的最小数量 ）maxSkew之间的最大允许差异。例如有 3 个 Zone，分别具有 2、4 和 5 个匹配的 pod，则全局最小值为 2
- 当 whenUnsatisfiable: ScheduleAnyway，scheduler 会为减少倾斜的拓扑提供更高的优先级。
minDomains：optional，符合条件的域的最小数量。
- 如果不指定该选项 minDomains，则约束的行为 minDomains: 1 。
- minDomains必须大于 0。minDomains与 whenUnsatisfiable 一起时为whenUnsatisfiable: DoNotSchedule。
topologyKey：Node label的key，如果多个Node都使用了这个lable key那么 scheduler 将这些 Node 看作为相同的拓扑域。
whenUnsatisfiable：当 Pod 不满足分布的约束时，怎么去处理
- DoNotSchedule（默认）不要调度。
- ScheduleAnyway仍然调度它，同时优先考虑最小化倾斜节点
labelSelector：查找匹配的 Pod label选择器的node进行技术，以计算Pod如何分布在拓扑域中

对于拓扑域的理解

对于拓扑域，官方是这么说明的，假设有一个带有以下lable的 4 节点集群：

NAME    STATUS   ROLES    AGE     VERSION   LABELS
node1   Ready    <none>   4m26s   v1.16.0   node=node1,zone=zoneA
node2   Ready    <none>   3m58s   v1.16.0   node=node2,zone=zoneA
node3   Ready    <none>   3m17s   v1.16.0   node=node3,zone=zoneB
node4   Ready    <none>   2m43s   v1.16.0   node=node4,zone=zoneB

那么集群拓扑如图：

图1：集群拓扑图 Source：https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/

假设一个 4 节点集群，其中 3个label被标记为foo: bar的 Pod 分别位于Node1、Node2 和 Node3：

图2：集群拓扑图 Source：https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/

这种情况下，新部署一个Pod，并希望新Pod与现有Pod跨 Zone均匀分布，资源清单文件如下：

kind: Pod
apiVersion: v1
metadata:
  name: mypod
  labels:
    foo: bar
spec:
  topologySpreadConstraints:
  - maxSkew: 1
    topologyKey: zone
    whenUnsatisfiable: DoNotSchedule
    labelSelector:
      matchLabels:
        foo: bar
  containers:
  - name: pause
    image: k8s.gcr.io/pause:3.1

这个清单对于拓扑域来说，topologyKey: zone 表示对Pod均匀分布仅应用于已标记的节点（如 foo: bar），将会跳过没有标签的节点（如zone: <any value>）。如果 scheduler 找不到满足约束的方法，whenUnsatisfiable: DoNotSchedule 设置的策略则是 scheduler 对新部署的Pod保持 Pendding

如果此时 scheduler 将新Pod 调度至 $Zone_A$，此时Pod分布在拓扑域间为 $[3,1]$ ，而 maxSkew 配置的值是1。此时倾斜值为 $Zone_A - Zone_B = 3-1=2$，不满足 maxSkew=1，故这个Pod只能被调度到 $Zone_B$。

此时Pod调度拓扑图为图3或图4

图3：集群拓扑图 Source：https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/

图4：集群拓扑图 Source：https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/

如果需要将Pod调度到 $Zone_A$ ,可以按照如下方式进行：

修改 maxSkew=2
修改 topologyKey: node 而不是 Zone ，这种模式下可以将 Pod 均匀分布在Node而不是Zone之间。
修改 whenUnsatisfiable: DoNotSchedule 为 whenUnsatisfiable: ScheduleAnyway 确保新的Pod始终可被调度

下面再通过一个例子增强对拓扑域了解

多拓扑约束

设拥有一个 4 节点集群，其中 3 个现有 Pod 标记 foo: bar分别位于 node1、node2 和 node3

图5：集群拓扑图 Source：https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/

部署的资源清单如下：可以看出拓扑分布约束配置了多个

kind: Pod
apiVersion: v1
metadata:
  name: mypod
  labels:
    foo: bar
spec:
  topologySpreadConstraints:
  - maxSkew: 1
    topologyKey: zone
    whenUnsatisfiable: DoNotSchedule
    labelSelector:
      matchLabels:
        foo: bar
  - maxSkew: 1
    topologyKey: node
    whenUnsatisfiable: DoNotSchedule
    labelSelector:
      matchLabels:
        foo: bar
  containers:
  - name: pause
    image: k8s.gcr.io/pause:3.1

在这种情况下，为了匹配第一个约束条件，新Pod 只能放置在 $Zone_B$ ；而就第二个约束条件，新Pod只能调度到 node4。在这种配置多约束条件下， scheduler 只考虑满足所有约束的值，因此唯一有效的是 node4。

如何为集群设置一个默认拓扑域约束

默认情况下，拓扑域约束也作 scheduler 的为 scheduler configurtion 中的一部分参数，这也意味着，可以通过profile为整个集群级别指定一个默认的拓扑域调度约束，

apiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration

profiles:
  - schedulerName: default-scheduler
    pluginConfig:
      - name: PodTopologySpread
        args:
          defaultConstraints:
            - maxSkew: 1
              topologyKey: topology.kubernetes.io/zone
              whenUnsatisfiable: ScheduleAnyway
          defaultingType: List

默认约束策略

如果在没有配置集群级别的约束策略时，kube-scheduler 内部 topologyspread 插件提供了一个默认的拓扑约束策略，大致上如下列清单所示

defaultConstraints:
  - maxSkew: 3
    topologyKey: "kubernetes.io/hostname"
    whenUnsatisfiable: ScheduleAnyway
  - maxSkew: 5
    topologyKey: "topology.kubernetes.io/zone"
    whenUnsatisfiable: ScheduleAnyway

上述清单中内容可以在 pkg\scheduler\framework\plugins\podtopologyspread\plugin.go

var systemDefaultConstraints = []v1.TopologySpreadConstraint{
    {
        TopologyKey:       v1.LabelHostname,
        WhenUnsatisfiable: v1.ScheduleAnyway,
        MaxSkew:           3,
    },
    {
        TopologyKey:       v1.LabelTopologyZone,
        WhenUnsatisfiable: v1.ScheduleAnyway,
        MaxSkew:           5,
    },
}

可以通过在配置文件中留空，来禁用默认配置

defaultConstraints: []
defaultingType: List

apiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration

profiles:
  - schedulerName: default-scheduler
    pluginConfig:
      - name: PodTopologySpread
        args:
          defaultConstraints: []
          defaultingType: List

通过源码学习Topology

podtopologyspread 实现了4种扩展点方法，包含 filter 和 score

PreFilter

可以看到 PreFilter 的核心为 calPreFilterState

func (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
    s, err := pl.calPreFilterState(ctx, pod)
    if err != nil {
        return nil, framework.AsStatus(err)
    }
    cycleState.Write(preFilterStateKey, s)
    return nil, nil
}

calPreFilterState 主要功能是用在计算如何在拓扑域中分布Pod，首先看段代码时，需要掌握下属几个概念

preFilterState
criticalPaths
update

func (pl *PodTopologySpread) calPreFilterState(ctx context.Context, pod *v1.Pod) (*preFilterState, error) {
    // 获取Node
    allNodes, err := pl.sharedLister.NodeInfos().List()
    if err != nil {
        return nil, fmt.Errorf("listing NodeInfos: %w", err)
    }
    var constraints []topologySpreadConstraint
    if len(pod.Spec.TopologySpreadConstraints) > 0 {
        // 这里会构建出TopologySpreadConstraints，因为约束是不确定的
        constraints, err = filterTopologySpreadConstraints(
            pod.Spec.TopologySpreadConstraints,
            v1.DoNotSchedule,
            pl.enableMinDomainsInPodTopologySpread,
            pl.enableNodeInclusionPolicyInPodTopologySpread,
        )
        if err != nil {
            return nil, fmt.Errorf("obtaining pod's hard topology spread constraints: %w", err)
        }
    } else {
        // buildDefaultConstraints使用".DefaultConstraints"与pod匹配的
        // service、replication controllers、replica sets 
        // 和stateful sets的选择器为pod构建一个约束。
        constraints, err = pl.buildDefaultConstraints(pod, v1.DoNotSchedule)
        if err != nil {
            return nil, fmt.Errorf("setting default hard topology spread constraints: %w", err)
        }
    }
    if len(constraints) == 0 { // 如果是空的，则返回空preFilterState
        return &preFilterState{}, nil
    }
    // 初始化一个 preFilterState 状态
    s := preFilterState{
        Constraints:          constraints,
        TpKeyToCriticalPaths: make(map[string]*criticalPaths, len(constraints)),
        TpPairToMatchNum:     make(map[topologyPair]int, sizeHeuristic(len(allNodes), constraints)),
    }
    // 根据node统计拓扑域数量
    tpCountsByNode := make([]map[topologyPair]int, len(allNodes))
    // 获取pod亲和度配置
    requiredNodeAffinity := nodeaffinity.GetRequiredNodeAffinity(pod)
    processNode := func(i int) {
        nodeInfo := allNodes[i]
        node := nodeInfo.Node()
        if node == nil {
            klog.ErrorS(nil, "Node not found")
            return
        }
        // 通过spreading去过滤node以用作filters，错误解析以向后兼容
        if !pl.enableNodeInclusionPolicyInPodTopologySpread {
            if match, _ := requiredNodeAffinity.Match(node); !match {
                return
            }
        }

        // 确保node的lable 包含topologyKeys定义的值
        if !nodeLabelsMatchSpreadConstraints(node.Labels, constraints) {
            return
        }

        tpCounts := make(map[topologyPair]int, len(constraints))
        for _, c := range constraints { // 对应的约束列表
            if pl.enableNodeInclusionPolicyInPodTopologySpread &&
                !c.matchNodeInclusionPolicies(pod, node, requiredNodeAffinity) {
                continue
            }
            // 构建出 topologyPair 以key value形式，
            // 通常情况下TopologyKey属于什么类型的拓扑
            //  node.Labels[c.TopologyKey] 则是属于这个拓扑中那个子域
            pair := topologyPair{key: c.TopologyKey, value: node.Labels[c.TopologyKey]}
            // 计算与标签选择器相匹配的pod有多少个
            count := countPodsMatchSelector(nodeInfo.Pods, c.Selector, pod.Namespace)
            tpCounts[pair] = count
        }
        tpCountsByNode[i] = tpCounts // 最终形成的拓扑结构
    }
    // 执行上面的定义的processNode，执行的数量就是node的数量
    pl.parallelizer.Until(ctx, len(allNodes), processNode)
    // 最后构建出 TpPairToMatchNum
    // 表示每个拓扑域中的每个子域各分布多少Pod，如图6所示
    for _, tpCounts := range tpCountsByNode {
        for tp, count := range tpCounts {
            s.TpPairToMatchNum[tp] += count
        }
    }
    if pl.enableMinDomainsInPodTopologySpread {
        // 根据状态进行构建 preFilterState
        s.TpKeyToDomainsNum = make(map[string]int, len(constraints))
        for tp := range s.TpPairToMatchNum {
            s.TpKeyToDomainsNum[tp.key]++
        }
    }

    // 计算最小匹配出的拓扑对
    for i := 0; i < len(constraints); i++ {
        key := constraints[i].TopologyKey
        s.TpKeyToCriticalPaths[key] = newCriticalPaths()
    }
    for pair, num := range s.TpPairToMatchNum {
        s.TpKeyToCriticalPaths[pair.key].update(pair.value, num)
    }

    return &s, nil // 返回的值则包含最小的分布
}

preFilterState

// preFilterState 是在PreFilter处计算并在Filter处使用。
// 它结合了 “TpKeyToCriticalPaths” 和 “TpPairToMatchNum” 来表示：
//（1）在每个分布约束上匹配最少pod的criticalPaths。 
// (2) 在每个分布约束上匹配的pod的数量。
// “nil preFilterState” 则表示没有设置（在PreFilter阶段）；
// empty “preFilterState”对象则表示它是一个合法的状态，并在PreFilter阶段设置。

type preFilterState struct {
    Constraints []topologySpreadConstraint

    // 这里记录2条关键路径而不是所有关键路径。 
    // criticalPaths[0].MatchNum 始终保存最小匹配数。 
    // criticalPaths[1].MatchNum 总是大于或等于criticalPaths[0].MatchNum，但不能保证是第二个最小匹配数。
    TpKeyToCriticalPaths map[string]*criticalPaths

    // TpKeyToDomainsNum 以 “topologyKey” 作为key ，并以zone的数量作为值。
    TpKeyToDomainsNum map[string]int

    // TpPairToMatchNum 以 “topologyPair作为key” ，并以匹配到pod的数量作为value。
    TpPairToMatchNum map[topologyPair]int
}

criticalPaths

// [2]criticalPath能够工作的原因是基于当前抢占算法的实现，特别是以下两个事实
// 事实 1：只抢占同一节点上的Pod，而不是多个节点上的 Pod。
// 事实 2：每个节点在其抢占周期期间在“preFilterState”的单独副本上进行评估。如果我们计划转向更复杂的算法，例如“多个节点上的任意pod”时则需要重新考虑这种结构。
type criticalPaths [2]struct {
    // TopologyValue代表映射到拓扑键的拓扑值。
    TopologyValue string
    // MatchNum代表匹配到的pod数量
    MatchNum int
}

单元测试中的测试案例，具有两个约束条件的场景，通过表格来解析如下：

Node列表与标签如下表：

Node Name	🏷️Lable-zone	🏷️Lable-node
node-a	zone1	node-a
node-b	zone1	node-b
node-x	zone2	node-x
node-y	zone2	node-y

Pod列表与标签如下表：

Pod Name	Node	🏷️Label
p-a1	node-a	foo:
p-a2	node-a	foo:
p-b1	node-b	foo:
p-y1	node-y	foo:
p-y2	node-y	foo:
p-y3	node-y	foo:
p-y4	node-y	foo:

对应的拓扑约束

spec:
  topologySpreadConstraints:
  - MaxSkew: 1
    TopologyKey: zone
    labelSelector:
      matchLabels:
        foo: bar
    MinDomains: 1
    NodeAffinityPolicy: Honor
    NodeTaintsPolicy: Ignore
  - MaxSkew: 1
    TopologyKey: node
    labelSelector:
      matchLabels:
        foo: bar
    MinDomains: 1
    NodeAffinityPolicy: Honor
    NodeTaintsPolicy: Ignore

那么整个分布如下：

图6：具有两个场景的分布图

实现的测试代码如下

{
    name: "normal case with two spreadConstraints",
    pod: st.MakePod().Name("p").Label("foo", "").
    SpreadConstraint(1, "zone", v1.DoNotSchedule, fooSelector, nil, nil, nil).
    SpreadConstraint(1, "node", v1.DoNotSchedule, fooSelector, nil, nil, nil).
    Obj(),
    nodes: []*v1.Node{
        st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
        st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
        st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
        st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
    },
    existingPods: []*v1.Pod{
        st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
        st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
        st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
        st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
        st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
        st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
        st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Obj(),
    },
    want: &preFilterState{
        Constraints: []topologySpreadConstraint{
            {
                MaxSkew:            1,
                TopologyKey:        "zone",
                Selector:           mustConvertLabelSelectorAsSelector(t, fooSelector),
                MinDomains:         1,
                NodeAffinityPolicy: v1.NodeInclusionPolicyHonor,
                NodeTaintsPolicy:   v1.NodeInclusionPolicyIgnore,
            },
            {
                MaxSkew:            1,
                TopologyKey:        "node",
                Selector:           mustConvertLabelSelectorAsSelector(t, fooSelector),
                MinDomains:         1,
                NodeAffinityPolicy: v1.NodeInclusionPolicyHonor,
                NodeTaintsPolicy:   v1.NodeInclusionPolicyIgnore,
            },
        },
        TpKeyToCriticalPaths: map[string]*criticalPaths{
            "zone": {{"zone1", 3}, {"zone2", 4}},
            "node": {{"node-x", 0}, {"node-b", 1}},
        },
        for pair, num := range s.TpPairToMatchNum {
        s.TpKeyToCriticalPaths[pair.key].update(pair.value, num)
    }
        TpPairToMatchNum: map[topologyPair]int{
            {key: "zone", value: "zone1"}:  3,
            {key: "zone", value: "zone2"}:  4,
            {key: "node", value: "node-a"}: 2,
            {key: "node", value: "node-b"}: 1,
            {key: "node", value: "node-x"}: 0,
            {key: "node", value: "node-y"}: 4,
        },
    },
},

update

update 函数实际上时用于计算 criticalPaths 中的第一位始终保持为是一个最小Pod匹配值

func (p *criticalPaths) update(tpVal string, num int) {
    // first verify if `tpVal` exists or not
    i := -1
    if tpVal == p[0].TopologyValue {
        i = 0
    } else if tpVal == p[1].TopologyValue {
        i = 1
    }

    if i >= 0 {
        // `tpVal` 表示已经存在
        p[i].MatchNum = num
        if p[0].MatchNum > p[1].MatchNum {
            // swap paths[0] and paths[1]
            p[0], p[1] = p[1], p[0]
        }
    } else {
        // `tpVal` 表示不存在，如一个新初始化的值
        // num对应子域分布的pod
        // 说明第一个元素不是最小的，则作为交换
        if num < p[0].MatchNum {
            // update paths[1] with paths[0]
            p[1] = p[0]
            // update paths[0]
            p[0].TopologyValue, p[0].MatchNum = tpVal, num
        } else if num < p[1].MatchNum {
            // 如果小于 paths[1]，则更新它，永远保证元素0是最小，1是次小的
            p[1].TopologyValue, p[1].MatchNum = tpVal, num
        }
    }
}

综合来讲 Prefilter 主要做的工作是。循环所有的节点，先根据 NodeAffinity 或者 NodeSelector 进行过滤，然后根据约束中定义的 topologyKeys （拓扑划分的依据）来选择节点。

接下来会计算出每个拓扑域下的拓扑对（可以理解为子域）匹配的 Pod 数量，存入 TpPairToMatchNum 中，最后就是要把所有约束中匹配的 Pod 数量最小（第二小）匹配出来的路径（代码是这么定义的，理解上可以看作是分布图）放入 TpKeyToCriticalPaths 中保存起来。整个 preFilterState 保存下来传递到后续的 filter 插件中使用。

Filter

在 preFilter 中最后的计算结果会保存在 CycleState 中

cycleState.Write(preFilterStateKey, s)

Filter 主要是从 PreFilter 处理的过程中拿到状态 preFilterState，然后看下每个拓扑约束中的 MaxSkew 是否合法，具体的计算公式为：$matchNum + selfMatchNum - minMatchNum$

matchNum：Prefilter 中计算出的对应的拓扑分布数量，可以在Prefilter中参考对应的内容
- if tpCount, ok := s.TpPairToMatchNum[pair]; ok {
selfMatchNum：匹配到label的数量，匹配到则是1，否则为0
minMatchNum：获的 Prefilter 中计算出来的最小匹配的值

func (pl *PodTopologySpread) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
    node := nodeInfo.Node()
    if node == nil {
        return framework.AsStatus(fmt.Errorf("node not found"))
    }
    // 拿到 prefilter处理的s，即preFilterState
    s, err := getPreFilterState(cycleState)
    if err != nil {
        return framework.AsStatus(err)
    }

    // 一个 空类型的 preFilterState是合法的，这种情况下将容忍每一个被调度的 Pod
    if len(s.Constraints) == 0 {
        return nil
    }

    podLabelSet := labels.Set(pod.Labels) // 设置标签
    for _, c := range s.Constraints { // 因为拓扑约束允许多个所以
        tpKey := c.TopologyKey
        tpVal, ok := node.Labels[c.TopologyKey]
        if !ok {
            klog.V(5).InfoS("Node doesn't have required label", "node", klog.KObj(node), "label", tpKey)
            return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReasonNodeLabelNotMatch)
        }

        // 判断标准
        // 现有的匹配数量 + 子匹配（1|0） - 全局minimum <= maxSkew
        minMatchNum, err := s.minMatchNum(tpKey, c.MinDomains, pl.enableMinDomainsInPodTopologySpread)
        if err != nil {
            klog.ErrorS(err, "Internal error occurred while retrieving value precalculated in PreFilter", "topologyKey", tpKey, "paths", s.TpKeyToCriticalPaths)
            continue
        }

        selfMatchNum := 0
        if c.Selector.Matches(podLabelSet) {
            selfMatchNum = 1
        }

        pair := topologyPair{key: tpKey, value: tpVal}
        matchNum := 0
        if tpCount, ok := s.TpPairToMatchNum[pair]; ok {
            matchNum = tpCount
        }
        skew := matchNum + selfMatchNum - minMatchNum
        if skew > int(c.MaxSkew) {
            klog.V(5).InfoS("Node failed spreadConstraint: matchNum + selfMatchNum - minMatchNum > maxSkew", "node", klog.KObj(node), "topologyKey", tpKey, "matchNum", matchNum, "selfMatchNum", selfMatchNum, "minMatchNum", minMatchNum, "maxSkew", c.MaxSkew)
            return framework.NewStatus(framework.Unschedulable, ErrReasonConstraintsNotMatch)
        }
    }

    return nil
}

minMatchNum

// minMatchNum用于计算 倾斜的全局最小值，同时考虑 MinDomains。
func (s *preFilterState) minMatchNum(tpKey string, minDomains int32, enableMinDomainsInPodTopologySpread bool) (int, error) {
    paths, ok := s.TpKeyToCriticalPaths[tpKey]
    if !ok {
        return 0, fmt.Errorf("failed to retrieve path by topology key")
    }
    // 通常来说最小值是第一个
    minMatchNum := paths[0].MatchNum
    if !enableMinDomainsInPodTopologySpread { // 就是plugin的配置的 enableMinDomainsInPodTopologySpread
        return minMatchNum, nil
    }

    domainsNum, ok := s.TpKeyToDomainsNum[tpKey]
    if !ok {
        return 0, fmt.Errorf("failed to retrieve the number of domains by topology key")
    }

    if domainsNum < int(minDomains) {
        // 当有匹配拓扑键的符合条件的域的数量小于 配置的"minDomains"(每个约束条件的这个配置) 时，
        //它将全局“minimum” 设置为0。
        // 因为minimum默认就为1，如果他小于1，就让他为0
        minMatchNum = 0
    }

    return minMatchNum, nil
}

PreScore

与 Filter 类似， PreScore 也是类似 PreFilter 的构成。 initPreScoreState 来完成过滤。

有了 PreFilter 基础后，对于 Score 来说大同小异

func (pl *PodTopologySpread) PreScore(
    ctx context.Context,
    cycleState *framework.CycleState,
    pod *v1.Pod,
    filteredNodes []*v1.Node,
) *framework.Status {
    allNodes, err := pl.sharedLister.NodeInfos().List()
    if err != nil {
        return framework.AsStatus(fmt.Errorf("getting all nodes: %w", err))
    }

    if len(filteredNodes) == 0 || len(allNodes) == 0 {
        // No nodes to score.
        return nil
    }

    state := &preScoreState{
        IgnoredNodes:            sets.NewString(),
        TopologyPairToPodCounts: make(map[topologyPair]*int64),
    }
    // Only require that nodes have all the topology labels if using
    // non-system-default spreading rules. This allows nodes that don't have a
    // zone label to still have hostname spreading.
    // 如果使用非系统默认分布规则，则仅要求节点具有所有拓扑标签。
    // 这将允许没有zone标签的节点仍然具有hostname分布。
    requireAllTopologies := len(pod.Spec.TopologySpreadConstraints) > 0 || !pl.systemDefaulted
    err = pl.initPreScoreState(state, pod, filteredNodes, requireAllTopologies)
    if err != nil {
        return framework.AsStatus(fmt.Errorf("calculating preScoreState: %w", err))
    }

    // return if incoming pod doesn't have soft topology spread Constraints.
    if len(state.Constraints) == 0 {
        cycleState.Write(preScoreStateKey, state)
        return nil
    }

    // Ignore parsing errors for backwards compatibility.
    requiredNodeAffinity := nodeaffinity.GetRequiredNodeAffinity(pod)
    processAllNode := func(i int) {
        nodeInfo := allNodes[i]
        node := nodeInfo.Node()
        if node == nil {
            return
        }

        if !pl.enableNodeInclusionPolicyInPodTopologySpread {
            // `node` should satisfy incoming pod's NodeSelector/NodeAffinity
            if match, _ := requiredNodeAffinity.Match(node); !match {
                return
            }
        }

        // All topologyKeys need to be present in `node`
        if requireAllTopologies && !nodeLabelsMatchSpreadConstraints(node.Labels, state.Constraints) {
            return
        }

        for _, c := range state.Constraints {
            if pl.enableNodeInclusionPolicyInPodTopologySpread &&
                !c.matchNodeInclusionPolicies(pod, node, requiredNodeAffinity) {
                continue
            }

            pair := topologyPair{key: c.TopologyKey, value: node.Labels[c.TopologyKey]}
            // If current topology pair is not associated with any candidate node,
            // continue to avoid unnecessary calculation.
            // Per-node counts are also skipped, as they are done during Score.
            tpCount := state.TopologyPairToPodCounts[pair]
            if tpCount == nil {
                continue
            }
            count := countPodsMatchSelector(nodeInfo.Pods, c.Selector, pod.Namespace)
            atomic.AddInt64(tpCount, int64(count))
        }
    }
    pl.parallelizer.Until(ctx, len(allNodes), processAllNode)
    // 保存状态给后面sorce调用
    cycleState.Write(preScoreStateKey, state)
    return nil
}

与Filter中Update使用的函数一样，这里也会到这一步，这里会构建出TopologySpreadConstraints，因为约束是不确定的

func filterTopologySpreadConstraints(constraints []v1.TopologySpreadConstraint, action v1.UnsatisfiableConstraintAction, enableMinDomainsInPodTopologySpread, enableNodeInclusionPolicyInPodTopologySpread bool) ([]topologySpreadConstraint, error) {
    var result []topologySpreadConstraint
    for _, c := range constraints {
        if c.WhenUnsatisfiable == action { // 始终调度时
            selector, err := metav1.LabelSelectorAsSelector(c.LabelSelector)
            if err != nil {
                return nil, err
            }
            tsc := topologySpreadConstraint{
                MaxSkew:            c.MaxSkew,
                TopologyKey:        c.TopologyKey,
                Selector:           selector,
                MinDomains:         1,                            // If MinDomains is nil, we treat MinDomains as 1.
                NodeAffinityPolicy: v1.NodeInclusionPolicyHonor,  // If NodeAffinityPolicy is nil, we treat NodeAffinityPolicy as "Honor".
                NodeTaintsPolicy:   v1.NodeInclusionPolicyIgnore, // If NodeTaintsPolicy is nil, we treat NodeTaintsPolicy as "Ignore".
            }
            if enableMinDomainsInPodTopologySpread && c.MinDomains != nil {
                tsc.MinDomains = *c.MinDomains
            }
            if enableNodeInclusionPolicyInPodTopologySpread {
                if c.NodeAffinityPolicy != nil {
                    tsc.NodeAffinityPolicy = *c.NodeAffinityPolicy
                }
                if c.NodeTaintsPolicy != nil {
                    tsc.NodeTaintsPolicy = *c.NodeTaintsPolicy
                }
            }
            result = append(result, tsc)
        }
    }
    return result, nil
}

Score

// 在分数扩展点调用分数。该函数返回的“score”是 `nodeName` 上匹配的 pod 数量，稍后会进行归一化。
func (pl *PodTopologySpread) Score(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, err := pl.sharedLister.NodeInfos().Get(nodeName)
    if err != nil {
        return 0, framework.AsStatus(fmt.Errorf("getting node %q from Snapshot: %w", nodeName, err))
    }

    node := nodeInfo.Node()
    s, err := getPreScoreState(cycleState)
    if err != nil {
        return 0, framework.AsStatus(err)
    }

    // Return if the node is not qualified.
    if s.IgnoredNodes.Has(node.Name) {
        return 0, nil
    }

    // 对于每个当前的 <pair>，当前节点获得 <matchSum> 的信用分。
    // 计算 <matchSum>总和 并将其作为该节点的分数返回。
    var score float64
    for i, c := range s.Constraints {
        if tpVal, ok := node.Labels[c.TopologyKey]; ok {
            var cnt int64
            if c.TopologyKey == v1.LabelHostname {
                cnt = int64(countPodsMatchSelector(nodeInfo.Pods, c.Selector, pod.Namespace))
            } else {
                pair := topologyPair{key: c.TopologyKey, value: tpVal}
                cnt = *s.TopologyPairToPodCounts[pair]
            }
            score += scoreForCount(cnt, c.MaxSkew, s.TopologyNormalizingWeight[i])
        }
    }
    return int64(math.Round(score)), nil
}

在 Framework 中会运行 ScoreExtension ，即 NormalizeScore

// Run NormalizeScore method for each ScorePlugin in parallel.
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
    pl := f.scorePlugins[index]
    nodeScoreList := pluginToNodeScores[pl.Name()]
    if pl.ScoreExtensions() == nil {
        return
    }
    status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
    if !status.IsSuccess() {
        err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
        errCh.SendErrorWithCancel(err, cancel)
        return
    }
})
if err := errCh.ReceiveError(); err != nil {
    return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
}

NormalizeScore 会为所有的node根据之前计算出的权重进行打分

func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
    s, err := getPreScoreState(cycleState)
    if err != nil {
        return framework.AsStatus(err)
    }
    if s == nil {
        return nil
    }

    // 计算 <minScore> 和 <maxScore>
    var minScore int64 = math.MaxInt64
    var maxScore int64
    for i, score := range scores {
        // it's mandatory to check if <score.Name> is present in m.IgnoredNodes
        if s.IgnoredNodes.Has(score.Name) {
            scores[i].Score = invalidScore
            continue
        }
        if score.Score < minScore {
            minScore = score.Score
        }
        if score.Score > maxScore {
            maxScore = score.Score
        }
    }

    for i := range scores {
        if scores[i].Score == invalidScore {
            scores[i].Score = 0
            continue
        }
        if maxScore == 0 {
            scores[i].Score = framework.MaxNodeScore
            continue
        }
        s := scores[i].Score
        scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore
    }
    return nil
}

到此，对于pod拓扑插件功能大概可以明了了，

Filter 部分（PreFilter，Filter）完成拓扑对(Topology Pair)划分
Score部分（PreScore, Score , NormalizeScore ）主要是对拓扑对（可以理解为拓扑结构划分）来选择一个最适合的pod的节点（即分数最优的节点）

而在 scoring_test.go 给了很多用例，可以更深入的了解这部分算法

Reference

^[1] scheduling code hierarchy

^[2] scheduler algorithm

^[3] in tree VS out of tree volume plugins

^[4] scheduler_framework_plugins

^[5] scheduling config

^[6] topology spread constraints

Kubernetes调度

cylon — Sun, 31 Jul 2022 14:32:42 +0000

[toc]

Overview

kube-scheduler 是kubernetes控制平面的核心组件，其默认行为是将 pod 分配给节点，同时平衡Pod与Node间的资源利用率。通俗来讲就是 kube-scheduler 在运行在控制平面，并将工作负载分配给 Kubernetes 集群。

本文将深入 Kubernetes 调度的使用，包含：”一般调度”，”亲和度“，“污点与容忍的调度驱逐”。最后会分析下 Scheduler Performance Tuning，即微调scheduler的参数来适应集群。

简单的调度

NodeName ^[1]

最简单的调度可以指定一个 NodeName 字段，使Pod可以运行在对应的节点上。如下列资源清单所示

apiVersion: v1
kind: Pod
metadata:
  name: netpod
spec:
  containers:
  - name: netbox
    image: cylonchau/netbox
  nodeName: node01

通过上面的资源清单Pod最终会在 node01上运行。这种情况下也会存在很多的弊端，如资源节点不足，未知的nodename都会影响到Pod的正常工作，通常情况下，这种方式是不推荐的。

$ kubectl describe pods netpod 
Name:         netpod
Namespace:    default

    ...

QoS Class:       BestEffort
Node-Selectors:  <none>
Tolerations:     node.kubernetes.io/not-ready:NoExecute for 300s
                 node.kubernetes.io/unreachable:NoExecute for 300s
Events:
  Type    Reason   Age   From     Message
  ----    ------   ----  ----     -------
  Normal  Pulling  86s   kubelet  Pulling image "cylonchau/netbox"
  Normal  Pulled   17s   kubelet  Successfully pulled image "cylonchau/netbox"
  Normal  Created  17s   kubelet  Created container netbox
  Normal  Started  17s   kubelet  Started container netbox



$ kubectl get pods netpod  -o wide
NAME     READY   STATUS    RESTARTS   AGE   IP            NODE     NOMINATED NODE   READINESS GATES
netpod   1/1     Running   0          48m   192.168.0.3   node01   <none>           <none>

通过上面的输出可以确定，通过 NodeName 方式是不经过 scheduler 调度的

nodeSelector ^[2]

label 是 kubernetes中一个很重要的概念，通常情况下，每一个工作节点都被赋予多组 label ,可以通过命令查看对应的 label 。

$ kubectl get node node01 --show-labels
NAME     STATUS   ROLES    AGE   VERSION    LABELS
node01   Ready    <none>   15h   v1.18.20   beta.kubernetes.io/arch=amd64,beta.kubernetes.io/os=linux,kubernetes.io/arch=amd64,kubernetes.io/hostname=node01,kubernetes.io/os=linux

而 nodeSelector 就是根据这些 label ，来选择具有特定一个或多个标签的节点。例如，如果需要在一组特定的节点上运行pod，可以设置在 “PodSpec” 中定义nodeSelector 为一组键值对：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-nodeselector
spec:
  selector:
    matchLabels:
      app: netpod
  replicas: 2 
  template:
    metadata:
      labels:
        app: netpod
    spec:
      containers:
      - name: netbox
        image: cylonchau/netbox
      nodeSelector: 
        beta.kubernetes.io/os: linux

对于上面的pod来讲，Kubernetes Scheduler 会找到带有 beta.kubernetes.io/os: linux标签的节点。对于更多kubernetes内置的标签，可以参考 ^[3]

对于标签选择器来说，最终会分布在具有标签的节点上

kubectl describe pod netpod-nodeselector-69fdb567d8-lcnv6 
Name:         netpod-nodeselector-69fdb567d8-lcnv6
Namespace:    default

    ...

QoS Class:       BestEffort
Node-Selectors:  beta.kubernetes.io/os=linux
Tolerations:     node.kubernetes.io/not-ready:NoExecute for 300s
                 node.kubernetes.io/unreachable:NoExecute for 300s
Events:
  Type    Reason     Age    From               Message
  ----    ------     ----   ----               -------
  Normal  Scheduled  8m18s  default-scheduler  Successfully assigned default/netpod-nodeselector-69fdb567d8-lcnv6 to node01
  Normal  Pulling    8m17s  kubelet            Pulling image "cylonchau/netbox"
  Normal  Pulled     7m25s  kubelet            Successfully pulled image "cylonchau/netbox"
  Normal  Created    7m25s  kubelet            Created container netbox
  Normal  Started    7m25s  kubelet            Started container netbox

节点亲和性 ^[4]

对于使用了调度功能的系统来说，亲和度（Affinity）是个很常见的概念，通常亲和度发生在并行（parallel）环境中；在这种环境下，亲和度提供了在一个节点上运行pod可能比在其他节点上运行更有效，而计算亲和度通常由多种条件组成。一般情况下，亲和度分为“软亲和与硬亲和

软亲和，Soft Affinity，是调度器尽可能将任务保持在同一个节点上。这只是一种尝试；如果不可行，则将进程迁移到另一个节点
硬亲和，Hard affinity，硬亲和度是强行将任务绑定到指定的节点上

而在kubernetes中也支持亲和度的概念，而亲和度是与 nodeSelector 配合形成的一个算法。其中硬亲和被定义为requiredDuringSchedulingIgnoredDuringExecution；软亲和被定义为 preferredDuringSchedulingIgnoredDuringExecution

硬亲和性（requiredDuringSchedulingIgnoredDuringExecution）：必须满足条件，否则调度程序无法调度 Pod。
软亲和性（preferredDuringSchedulingIgnoredDuringExecution）：scheduler 将查找符合条件的节点。如果没有满足要求的节点将忽略这条规则，scheduler 将仍会调度 Pod。

Node Affinity

Node Affinity参数说明

调度程序会更倾向于将 pod 调度到满足该字段指定的亲和性表达式的节点，但它可能会选择违反一个或多个表达式的节点。最优选的节点是权重总和最大的节点，即对于满足所有调度要求（资源请求、requiredDuringScheduling 亲和表达式等）的每个节点，通过迭代该字段的元素来计算总和如果节点匹配相应的matchExpressions，则将“权重”添加到总和中；具有最高和的节点是最优选的。

如果在调度时不满足该字段指定的亲和性要求，则不会将 Pod 调度到该节点上。如果在 pod 执行期间的某个时间点不再满足此字段指定的亲和性要求（例如，由于更新），系统可能会或可能不会尝试最终将 pod 从其节点中逐出。

affinity 范围应用于 Pod.Spec 下，参数如下：

nodeAffinity：node亲和度相关根配置
- preferredDuringSchedulingIgnoredDuringExecution：软亲和
  - preference (required)：选择器
    - matchExpressions：匹配表达式，标签可以指定部分
      - key (<string> -required-)：
      - operator (<string> -required-)：# 与一组 key-values的运算方式。
      - values (<[]string>)：
    - matchFields：匹配字段
      - key (<string> -required-)：
      - operator (<string> -required-)：# 与一组 key-values的运算方式。
      - values (<[]string>)：
  - weight (required)：范围为 1-100，具有最高和的节点是最优选的
- requiredDuringSchedulingIgnoredDuringExecution：硬亲和
  - nodeSelectorTerms
    - matchExpressions：
      - key (<string> -required-)：
      - operator (<string> -required-)：# 与一组 key-values的运算方式。
      - values (<[]string>)：
    - matchFields：
      - key (<string> -required-)：
      - operator (<string> -required-)：一组 key-values 的运算方式。
      - values (<[]string>)：

Notes: matchFields使用的是资源清单的字段（kubectl get node -o yaml），而matchExpressions匹配的是标签

Node Affinity示例

上面的介绍了解到了Kubernetes中相对与 nodeSelector可以更好表达复杂的调度需求：节点亲和性，使用PodSpec中的字段 .spec.affinity.nodeAffinity 指定相关 affinity 配置。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-nodeselector
spec:
  selector:
    matchLabels:
      app: netpod
  replicas: 2 
  template:
    metadata:
      labels:
        app: netpod
    spec:
      containers:
      - name: netbox
        image: cylonchau/netbox
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: app
                operator: In
                values:
                - test

上面的清单表明，当节点存在 app: test 标签时，会调度到对应的Node上，如果没有节点匹配这些条件也不要紧，会根据普通匹配进行调度。

当硬策略和软策略同时存在时的情况，根据设置的不同，硬策略优先级会高于软策略，哪怕软策略权重设置为100

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-nodeselector
spec:
  selector:
    matchLabels:
      app: netpod
  replicas: 2 
  template:
    metadata:
      labels:
        app: netpod
    spec:
      containers:
      - name: netbox
        image: cylonchau/netbox
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: app
                operator: In 
                values:
                - test
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: topology.kubernetes.io/zone
                operator: In
                values:
                - antarctica-east1
                - antarctica-west1

下面是报错信息

Warning  FailedScheduling  4s (x3 over 24s)  default-scheduler  0/2 nodes are available: 2 node(s) didn't match node selector.

Pod亲和性 ^[4]

pod亲和性和反亲和性是指根据节点上已运行的Pod的标签而不是Node标签来限制Pod可以在哪些节点上调度。例如：X 满足一个或多个运行 Y 的条件，这个时候 Pod满足在X中运行。其中 X 为拓扑域，Y 则是规则。

Notes：官方文档中不推荐pod亲和度在超过百个节点的集群中使用该功能 ^[5]

Pod亲和性配置

Pod亲和性和反亲和性与Node亲和性类似，affinity 范围应用于 Pod.Spec.podAffinity 下，这里不做重复复述，可以参考Node亲和性参数说明部分。

topologyKey，==不允许是空值==，该值将影响Pod部署的位置，影响范围为，与亲和性条件匹配的对应的节点中的什么拓扑，topologyKey的拓扑域由label标签决定。

除了 topologyKey 之外，还有标签选择器 labelSelector 与名称空间 namespaces 可以作为同级的替代选项

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-podaffinity
spec:
  selector:
    matchLabels:
      app: podaffinity
  replicas: 1 
  template:
    metadata:
      labels:
        app: podaffinity
    spec:
      containers:
      - name: podaffinity
        image: cylonchau/netbox
      affinity:
        podAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchExpressions:
              - key: app
                operator: In
                values:
                - netpod
            topologyKey: zone

如果没有Pod匹配到规则，则pending状态

Warning  FailedScheduling  59s (x2 over 59s)  default-scheduler  0/2 nodes are available: 2 node(s) didn't match pod affinity rules, 2 node(s) didn't match pod affinity/anti-affinity.

Pod Anti-Affinity

在某些场景下，部分节点不应该有很多资源，即某些节点不想被调度。例如监控运行节点由于其性质，不希望该节点上有很多资源，或者因节点配置的不同，配置较低节点不希望调度很多资源；在这种情况下，如果将符合预期之外的Pod调度过来会降低其托管业务的性能。这种情况下就需要 反亲和度（Anti-Affinity）来使Pod远离这组节点

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-podaffinity
spec:
  selector:
    matchLabels:
      app: podaffinity
  replicas: 1 
  template:
    metadata:
      labels:
        app: podaffinity
    spec:
      containers:
      - name: podaffinity
        image: cylonchau/netbox
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - netpod
              topologyKey: zone

Taints And Tolerations ^[6]

Taints

亲和度和反亲和度虽然可以阻止Pod在特定节点上运行，但还存在一个问题，就是亲和度和反亲和度需要声明运行的节点或者是不想运行的节点，如果忘记声明，还是会被调度到对应的Node上。Kubernetes还提供了一种驱逐Pod的方法，就是污点（Taints）与容忍（Tolerations）。

创建一个污点

kubectl taint nodes node1 key1=value1:NoSchedule

$ kubectl taint nodes mon01 role=monitoring:NoSchedule

删除一个污点，

kubectl taint nodes node1 key1=value1:NoSchedule-

除了 NoSchedule ，还有 PreferNoSchedule 与 NoExecute

PreferNoSchedule ：类似于软亲和性的属性，尽量去避免污点，但不是强制的。
NoExecute 表示，当Pod还没在节点上运行时，并且存在至少一个污点时生效，此时Pod不会被调度到该节点；当Pod已经运行在节点上时，并且存在至少一个污点时生效，Pod将会从节点上被驱逐。

Tolerations

当Node有污点时，在调度时会自动被排除。当调度在受污染的节点上执行Predicate部分时将失败，而容忍度则是使 pod 具有对该节点上的污点进行容忍，即拥有容忍度的Pod可以调度到有污点的节点之上。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: netpod-podaffinity
spec:
  selector:
    matchLabels:
      app: podaffinity
  replicas: 1 
  template:
    metadata:
      labels:
        app: podaffinity
    spec:
      containers:
      - name: podaffinity
        image: cylonchau/netbox
      tolerations:
      - key: "role"
        operator: "Equal"
        value: "monitoring"
        effect: "NoSchedule"
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - netpod
              topologyKey: zone

容忍度中存在一个特殊字段 TolerationSeconds ，表示容忍的时间，默认不会设置，即永远容忍污点。设置成0或者负数表示理解驱逐。==仅在污点为 NoExecute 时生效==

operator 属性有两个值 Exists 和 Equal

如果 operator 为 Exists，则无需 value 属性，因为判断的是有污点的情况下。
如果 operator 为 Equal，则表示 key 与 value 之间的关系是 $key=value$
空 key，并且operator为 Exists，将匹配到所有，即容忍所有污点
空 effect 匹配所有 effect ，即匹配所有污点；这种情况下加上条件的话，可以容忍所有类型的污点

驱逐 ^[7]

当污点设置为 NoExecute这种情况下会驱逐Pod，驱逐条件又如下几个：

不容忍污点的 pod 会立即被驱逐
容忍污点但未配置 tolerationSeconds 属性的会保持不变，即该节点与Pod保持绑定
容忍指定污点的 pod 并且配置了tolerationSeconds 属性，节点与Pod绑定状态仅在配置的时间内。

Kubernetes内置了一些污点，此时 Controller 会自动污染节点：

node.kubernetes.io/not-ready: Node故障。对应 NodeCondition 的Ready = False。
node.kubernetes.io/unreachable：Node控制器无法访问节点。对应 NodeCondition Ready= Unknown。
node.kubernetes.io/memory-pressure：Node内存压力。
node.kubernetes.io/disk-pressure：Node磁盘压力。
node.kubernetes.io/pid-pressure：Node有PID压力。
node.kubernetes.io/network-unavailable：Node网络不可用。
node.kubernetes.io/unschedulable：Node不可调度。

Notes：Kubernetes node.kubernetes.io/not-ready 属性和 node.kubernetes.io/unreachable 属性添加容差时效 tolerationSeconds=300。即在检测到其中问题后，Pod 将保持绑定5分钟。

优先级和抢占

kubernetes中也为Pod提供了优先级的机制，有了优先级机制就可以在并行系统中提供抢占机制，有了抢占机制后，当还未调度时，高优先级Pod会比低优先级Pod先被调度，在资源不足时，低优先级Pod可以被高优先级Pod驱逐。

优先级功能由 PriorityClasses 提供。PriorityClasses 是作为集群级别资源而不是命名空间级别资源，只是用来声明优先级级别。

value 作为优先级级别，数字越大优先级级别越高。而 name 是这个优先级的名称，与其他资源name值相似，值的内容需要符合DNS域名约束。

globalDefault 是集群内默认的优先级级别，仅只有一个 PriorityClass 可以设置为 true

apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: high-priority
value: 1000000
globalDefault: false
description: "This priority class should be used for XYZ service pods only."

Notes:

如果集群内不存在任何 PriorityClass ，则存在的Pod的优先级都为0

当对集群设置了 globalDefault=true 后，不会改变已经存在的 Pod 的优先级。仅对于 PriorityClass globalDefault=true 后创建的 Pod。

如果删除了 PriorityClass ，存在还是使用的这个 PriorityClass 的Pod保持不变，新创建的Pod无法使用这个 PriorityClass 。

非抢占

当 preemptionPolicy: Never 时，Pod不会抢占其他Pod，但不可调度时，会一直在调度队列中等待调度，直到满足要求才会被调度。==非抢占式pod仍可能被其他高优先级的pod抢占==

Notes：preemptionPolicy在Kubernetes v1.24 [stable]

preemptionPolicy 是作为非抢占的配置，默认参数为 PreemptLowerPriority；表示了允许高优先级Pod抢占低优先级Pod。如果 preemptionPolicy: Never，代表Pod是非抢占式的。

下列是一个非抢占式的配置样例

apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: high-priority-nonpreempting
value: 1000000
preemptionPolicy: Never
globalDefault: false
description: "This priority class will not cause other pods to be preempted."

当配置了优先级后，优先级准入控制器会使用 priorityClassName 中配置的对应的 PriorityClass 的 value值来填充当前Pod的优先级，如果没有找到对应抢占策略，则拒绝。

下面是在Pod中配置优先级的示例

apiVersion: v1
kind: Pod
metadata:
  name: nginx
  labels:
    env: test
spec:
  containers:
  - name: nginx
    image: nginx
    imagePullPolicy: IfNotPresent
  priorityClassName: high-priority

抢占

当创建Pod之后，Pod会进入队列并等待被调度。scheduler 从队列中选择一个 Pod 并尝试将其调度到一个节点上。如果没有找到满足 Pod 的所有指定要求的 Node，则为 Pending 的 Pod 触发抢占。当Pod在找合适的节点时，即试图抢占一个节点，会在这个Node中删除一个或多个优先级低于当前Pod的Pod，使当前Pod能够被调度到对应的Node上。当低优先级Pod被驱逐后，当前Pod可以被调度到该Node，这个过程被称为抢占 preemption。

而提供可驱逐资源的Node成为被提名Node（nominated Node），在当Pod抢占到一个Node时，其 nominatedNodeName 会被标注为这个 Node的名称，当然标注后也不一定，一定是被抢占到这个Node之上，例如，当前Pod在等待驱逐低优先级Pod的过程中，有其他节点变成可用节点 FN 时，这个时候Pod会被抢占到这个节点。

Reference

^[1] 创建一个会被调度到特定节点上的 Pod

^[2] nodeSelector

^[3] labels annotations taints

^[4] affinity and anti-affinity

^[5] inter pod affinity and anti affinity

^[6] taint and toleration

^[7] evictions

^[8] pod priority preemption

DEV Community: cylon

[open source] based-firewalld Linux iptables UI&firewalld UI distributed management platform

Best firewalld UI

Uranus Gateway

Project Show

Special feature

future

【github开源】Linux iptables 界面化 分布式 管理平台 go vue，底座firewalld

目前最优的firewalld前端

Uranus Gateway

环境的展示

特色功能

未来功能投票

kubernetes networking model

Overview

Underlay Network Model

什么是Underlay Network

kubernetes中的underlay network

flannel host-gw [1]

Calico BGP [2]

IPVLAN & MACVLAN [4]

multus

danm

Overlay Network Model

什么是Overlay

常见的网络隧道技术

IPIP

VxLAN

weave vxlan [3]

H3C HCL与WSL2共存

Background

安装过程

VirtualBox启用hyper-v支持 [2]

巧用Prometheus来扩展kubernetes调度器

Overview

kubernetes调度配置

Scheduler Configuration [1]

kubeSchedulerConfiguration使用

scheduler调度插件 [3]

如何扩展kube-scheduler [4]

定义入口

插件实现

定义插件结构体

实现对应的扩展点

实验：基于网络流量的调度 [7]

实验环境

实验开始

错误处理

定义插件API

定义扩展点

定义要传入的参数

配置调度器的参数

项目部署

验证结果

最小化安装debian11

Preparation

如何选择下载安装包

安装步骤

选择位置与键盘布局

设置主机名和域名

设置非ROOT用户名、账户和密码

设置时钟时区

对磁盘分区

Base System安装

离线安装

根据要求调整安装

完成Debian11最小化安装

深入解析kubernetes中的选举机制

Overview

Background

election in controller-runtime

leaselock

Get

Create

Update

RecordEvent

election workflow

summary

让你的Nginx支持分布式追踪opentracing

Background

【github开源】Linux iptables 界面化分布式管理平台 go vue，底座firewalld

flannel host-gw ^[1]

Calico BGP ^[2]

IPVLAN & MACVLAN ^[4]

weave vxlan ^[3]

VirtualBox启用hyper-v支持 ^[2]

Scheduler Configuration ^[1]

scheduler调度插件 ^[3]

如何扩展kube-scheduler ^[4]

实验：基于网络流量的调度 ^[7]

调度框架 ^[2]

扩展点 ^[3]

调度框架 ^[1]

插件 ^[4]

NodeResourcesFit ^[5]

Topology ^[6]

NodeName ^[1]

nodeSelector ^[2]