Use API

Concept

kubectl create、kubectl replace是命令式配置文件操作，kubectl apply是声明式API。

Practice

本地编写一个nginx-deployment.yaml文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx-deployment
  labels:
    app: nginx
spec:
  replicas: 3
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx:1.7.9
        ports:
        - containerPort: 80

使用kubectl apply命令来创建这个Deployment：

1
2
$ kubectl apply -f nginx-deployment.yaml
deployment.apps/nginx-deployment created

然后修改一下nginx-deployment.yaml里定义的镜像：

1
2
3
4
5
6
7
...
    spec:
      containers:
      - name: nginx
        image: nginx:1.21.6
        ports:
        - containerPort: 80

这时继续执行kubectl apply命令：

1
2
$ kubectl apply -f nginx-deployment.yaml
deployment.apps/nginx-deployment configured

这时，Kubernetes就会触发这个Deployment的滚动更新。

Problem

这样触发的更新与kubectl create再kubectl replace有什么区别呢？

kubectl replace的执行过程，是使用新的YAML文件中的API对象，替换原有的API对象；而kubectl apply，则是执行了一个对原有API对象的PATCH操作。

更进一步地，这意味着kube-apiserver在响应命令式请求（比如kubectl replace）对时候，一次只能处理一个写请求，否则会有产生冲突的可能。而对于声明式请求（比如kubectl apply）的时候，一次能处理多个写操作，并且具体Merge的能力。

Meaning

Istio

下面以Istio项目为例，学习声明式API在实际使用时的重要意义。

Istio官网：https://istio.io/latest/zh/

Server Mesh：https://www.servicemesher.com/istio-handbook/concepts/basic.html

Istio是一个开源的Service Mesh实现产品，实际上就是一个基于Kubernetes项目的微服务治理框架，它的架构非常清晰，如下所示：


在上面这个架构图中，不难看到Istio项目架构的核心所在。Istio最根本的组件，是运行在每一个应用Pod里的Envoy容器。

这个Envoy项目是Lyft公司推出的一个高性能C++网络代理，也是Lyft公司对Istio项目的唯一贡献。

而Istio项目，则把这个代理服务以sidecar容器的方式，运行在了每一个被治理的应用Pod中。Pod里的所有容器都共享同一个Network Namespace。所以，Envoy容器就能够通过配置Pod里面的iptables规则，把整个Pod的进出流量接管下来。

这时候，Istio的控制层（Control Plane）里的Pilot组件，就能够通过调用每个Envoy容器的API，对这个Envoy代理进行配置，从而实现微服务治理。

看个🌰：假设这个Istio架构图左边的Pod是已经在运行的应用，而右边的Pod则是我们刚刚上线的应用的新版本。这时候，Pilot通过调节这两Pod里的Enovy容器的配置，从而将90%的流量分配给旧版本的应用，将10%的流量分配给新版本应用，并且，还可以在后续的过程中随时调整。这样，一个典型的灰度发布的场景就完成了。比如，Istio可以调节这个流量从90%-10%，改到80%-20%，再到50%-50%，最后到0%-100%，就完成了这个灰度发布的过程。

更重要的是，在整个微服务治理的过程中，无论是对Envoy容器的部署，还是像上面这样对Envoy代理的配置，用户和应用都是完全“无感”的。

Istio项目使用的，是Kubernetes中一份非常重要的功能，叫做Dynamic Admission Controll。

在Kubernetes项目中，当一个Pod或者任何一个API对象被提交给APIServer之后，总有一些“初始化”性质的工作需要在它们被Kubernetes项目正式处理之前进行。比如，自动为所有Pod加上某些标签（Labels）。

而这个“初始化”操作的实现，借助的是一个叫作Admission的功能。它其实是Kubernetes项目里一组被称为Admission Controller的代码，可以选择性地被编译进APIServer中，在API对象创建之后会被立刻调用到。

但这就意味着，如果现在想要添加一些自己的规则到Adminssion Controller，就会比较困难。因为，这要求重新编译并重启APIServer。显然，这种使用方法对Istio来说，影响太大了。

所以，Kubernetes项目为我们提供了一种“热插拔”式的Admission机制，它就是Dynamic Admission Control，也叫作Initializer。

Example

比如，现在有如下所示的一个应用Pod：

1
2
3
4
5
6
7
8
9
10
11
apiVersion: v1
kind: Pod
metadata:
  name: myapp-pod
  labels:
    app: myapp
spec:
  containers:
  - name: myapp-container
    image: busybox
    command: ['sh', '-c', 'echo Hello Kubernetes! && sleep 3600']

可以看到，这个Pod里面只有一个用户容器，叫作：myapp-container。

接下来，Istio项目要做的，就是在这个Pod的YAML被提交给Kubernetes之后，在它对应的API对象里自动加上Envoy容器的配置，使这个对象变成如下所示的样子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
apiVersion: v1
kind: Pod
metadata:
  name: myapp-pod
  labels:
    app: myapp
spec:
  containers:
  - name: myapp-container
    image: busybox
    command: ['sh', '-c', 'echo Hello Kubernetes! && sleep 3600']
  - name: envoy
    image: lyft/envoy:845747b88f102c0fd262ab234308e9e22f693a1
    command: ["/usr/local/bin/envoy"]
    ...

可以看到，被Istio处理后的这个Pod里，除了用户自己定义的myapp-container容器之外，多出了一个叫作envoy的容器，它就是IStio要使用的Envoy代理。

Basic Principle

那么，Istio又是如何在用户完全不知情的前提下完成这个操作的呢？

Istio要做的，就是编写一个用来为Pod“自动注入”Envoy容器的Initializer。

首先，Istio会将这个Envoy容器本身的定义，以ConfigMap的方式保存在Kubernetes当中。这个ConfigMap（名叫：envoy-initializer）的定义如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
apiVersion: v1
kind: ConfigMap
metadata:
  name: envoy-initializer
data:
  config: |
    containers:
      - name: envoy
        image: lyft/envoy:845747db88f102c0fd262ab234308e9e22f693a1
        command: ["/usr/local/bin/envoy"]
        args:
          - "--concurrency 4"
          - "--config-path /etc/envoy/envoy.json"
          - "--mode serve"
        ports:
          - containerPort: 80
            protocol: TCP
        resources:
          limits:
            cpu: "1000m"
            memory: "512Mi"
          requests:
            cpu: "100m"
            memory: "64Mi"
        volumeMounts:
          - name: envoy-conf
            mountPath: /etc/envoy
    volumes:
      - name: envoy-conf
        configMap:
          name: envoy

这个ConfigMap的data部分，正是一个Pod对象的一部分定义，其中，我们可以看到Envoy容器对应的containers字段，以及一个用来声明Envoy配置文件的volumes字段。

不难想到，Initializer要做的工作，就是把这部分Envoy相关的字段，自动添加到用户提交的Pod的API对象里。可是，用户提交的Pod里本来就有containers字段和volumes字段，所以Kubernetes在处理这样的更新请求时，就必须使用类似于git merge这样的操作，才能将这两部分内容结合在一起。

所以说，在Initializer更新用户的Pod对象的时候，必须使用PATCH API来完成。而这种PATCH API，正是声明式API最主要的能力。

接下来，Istio将一个编写好的Initializer，作为一个Pod部署在Kubernetes中。这个Pod的定义非常简单，如下所示：

1
2
3
4
5
6
7
8
9
10
11
apiVersion: v1
kind: Pod
metadata:
  labels:
    app: envoy-initializer
  name: envoy-initializer
spec:
  containers:
    - name: envoy-initializer
      image: envoy-initializer:0.0.1
      imagePullPolicy: Always

可以看到，这个envoy-initializer使用的envoy-initializer:0.0.1镜像，就是一个事先编写好的“自定义控制器”（Custom Controller）。

一个Kubernetes Controller，实际上就是一个死循环：它不断地获取实际状态，然后与期望状态，并以此为依据决定下一步的操作。

而Initializer的控制器，不断获取到的实际状态，就是用户新创建的Pod。而它的“期望状态”，则是：这个Pod里被添加了Envoy容器的定义。

用Go语言的伪代码描述这个控制逻辑，如下所示：

1
2
3
4
5
6
7
8
9
for {
  // 获取新创建的Pod
  pod := client.GetLatestPod()
  // Diff一下，检查是否已经初始化过
  if !isInitialized(pod) {
    // 没有？那就来初始化一下
    doSomthing(pod)
  }
}
  • 如果这个Pod里面已经添加过Envoy容器，那么就“放过”这个Pod，进入下一个检查周期。
  • 而如果还没有添加过Envoy容器的话，它就要进行Initilize操作了，即：修改该Pod的API对象（doSomething函数）。

这时候，Istio要往这个Pod里合并的字段，正是我们之前保存在envoy-initializer这个ConfigMap里的数据（即：它的data字段的值）。

所以，在Initializer控制器的工作逻辑里，它首先会从APIServer中拿到这个ConfigMap：

1
2
3
func doSomthing(pod) {
  cm := client.Get(Config, "envoy-initializer")
}

然后，把这个ConfigMap里存储的containers和volumes字段，直接添加进一个空的Pod对象里：

1
2
3
4
5
6
7
func doSomething(pod) {
  cm := client.Get(Config, "envoy-initializer")
  
  newPod := Pod{}
  newPod.Spec.Containers = cm.Containers
  newPod.Spec.Volumes = cm.Volumes
}

现在，关键来了。Kubernetes的API库，提供了一个方法，可以直接使用新旧两个Pod对象，生成一个TwoWayMergePatch：

1
2
3
4
5
6
7
8
9
10
11
12
13
func doSomething(pod) {
  cm := client.Get(Config, "envoy-initializer")
  
  newPod := Pod{}
  newPod.Spec.Containers = cm.Containers
  newPod.Spec.Volumes = cm.Volumes
  
  // 生成patch数据
  patchBytes := strategicpatch.CreateTwoWayMergePatch(pod, newPod)
  
  // 发起PATCH请求，修改这个Pod对象
  client.Patch(pod.Name, patchBytes)
}

有了这个TwoWayMergePatch之后，Initializer的代码就可以使用这个patch的数据，调用Kubernetes的Client，发起一个PATCH请求。

这样，一个用户提交的Pod对象里，就会被自动加上Envoy容器相关的字段。

Other Method

当然，Kubernetes还允许通过配置，来指定要对什么样的资源进行这个Initialize操作，比如下面这个🌰：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
apiVersion: admissionregistration.k8s.io/v1alpha1
kind: InitializerConfiguration
metadata:
  name: envoy-config
initializers:
  // 这个名字必须至少包括两个 "."
  - name: envoy.initializer.kubernetes.io
    rules:
      - apiGroups:
          - "" // 前面说过， ""就是core API Group的意思
        apiVersions:
          - v1
        resources:
          - pods

这个配置，就意味着Kubernetes要对所有的Pod进行这个Initialize操作，并且，我们指定了负责这个操作的Initializer，名叫：envoy-initializer。

而一旦这个InitializerConfiguration被创建，Kubernetes就会把这个Initializer的名字，加在所有新创建的Pod的Metadata上，格式如下所示：

1
2
3
4
5
6
7
8
9
10
apiVersion: v1
kind: Pod
metadata:
  initializers:
    pending:
      - name: envoy.initializer.kubernetes.io
  name: myapp-pod
  labels:
    app: myapp
...

可以看到，每一个新创建的Pod，都会自动懈怠了metadata.initializer.pending的Metadata信息。

这个Metadata，正是接下来Initializer的控制器判断这个Pod有没有执行过自己所负责的初始化操作的重要依据（也就是前面伪代码中isInitialized()方法的含义）。

这也意味着，当你在Initializer里完成了要做的操作后，一定要记得将这个metadata.initializers.pending标志清除掉。这一点，在编写Initializer代码的时候一定要非常注意。

此外，除了上面的配置方法，还可以在具体的Pod的Annotation里添加一个如下所示的字段，从而声明要使用某个Initializer：

1
2
3
4
5
6
apiVersion: v1
kind: Pod
metadata
  annotations:
    "initializer.kubernetes.io/envoy": "true"
    ...

在这个Pod里，我们添加了一个Annotation，写明：initializer.kubernetes.io/envoy=true。这样，就会使用到我们前面所定义的envoy-initializer了。

Summary

Istio项目的核心，就是由无数个运行在应用Pod中的Envoy容器组成的服务代理网格。这也正是Service Mesh的含义。

而这个机制得以实现的原理，正是借助了Kubernetes能够对API对象进行在线更新的能力，这也正是Kubernetes“声明式API”的独特之处。

  • 首先，所谓“声明式”，指的就是我只需要提交一个定义好的API对象来“声明”，我所期望的状态是什么样子。
  • 其次，”声明式API“允许有多个API写端，以PATCH的方式对API对象进行修改，而无需关心本地原始YAML文件的内容。
  • 最后，也是最重要的，有了上述两个能力，Kubernetes项目才可以基于对API对象的增、删、改、查，在完全无需外界干预的情况下，完成对实际状态和期望状态的调谐（Reconcile）过程。

所以说，声明式API，才是Kubernetes项目编排能力“赖以生存”的核心所在。

无论是对sidecar容器的巧妙设计，还是对Initializer的合理利用，Istio项目的设计与实现，其实都依托于Kubernetes的声明式API和它所提供的各种编排能力。可以说，Istio是Kubernetes项目使用上的一位“集大成者”。

API’s secret

Design

一直好奇：当我把一个YAML文件提交给Kubernetes之后，它究竟是如何创建出一个API对象的呢？

在Kubernetes中，一个API对象在Etcd里的完整资源路径，是由Group（API组）、Version（API版本）和Resource（API资源类型）三个部分组成的。

通过这样的结构，整个Kubernetes里的所有API对象，实际上就可以用如下的树形结构表示出来：


在这张图里，可以清除地看到Kubernetes里API对象的组织方式，其实是层层递进的。

比如，现在要声明创建一个CronJob对象，那么YAML文件的开始部分会这么写：

1
2
3
apiVersion: batch/v2alpha1
kind: CronJob
...

在这个YAML文件中，“CronJob”就是这个API对象的资源类型（Resource），“batch”就是它的组（Group），“v2alpha1”就是它的版本（Version）。

当我们提交了这个YAML文件之后，Kubernetes就会把这个YAML文件里描述的内容，转换成Kubernetes里的一个CronJob对象。

Problem

那么，Kubernetes是如何对Resource、Group和Version进行解析，从而在Kubernetes项目里找到CronJob对象的定义呢？

首先，Kubernetes会匹配API对象的组。

需要明确的是，对于Kubernetes里的核心API对象，比如：Pod、Node等，是不需要Group的（即它们的Group是“”）。所以，对于这些API对象来说，Kubernetes会直接在/api这个层级进行下一步的匹配过程。

而对于CronJob等非核心API对象来说，Kubernetes就必须在/apis这个层级里查找它对应的Group，进而根据“batch”这个Group的名字，找到/apis/batch。

不难发现，这些API Group的分类是以对象功能为依据的，比如Job和Cronjob就都属于“batch”（离线业务）这个Group。

然后，Kubernetes会进一步匹配到API对象的版本号。

对于CronJob这个API对象来说，Kubernetes在batch这个Group下，匹配到的版本号就是v2alpha1。

在Kubernetes中，同一种API对象可以有多个版本，这正是Kubernetes进行API版本化管理的重要手段。这样，比如在CronJob的开发过程中，对于会影响到用户的变更就可以通过升级新版本号来处理，从而保证了向后兼容。

最后，Kubernetes会匹配API对象的资源类型。

在前面匹配到正确的版本之后，Kubernetes就知道要创建一个/apis/batch/v2alpha1下的CronJob对象。

这时候，APIServer就可以继续创建这个CronJob对象了。

流程图如下：


首先，当我们发起了创建CronJob的POST请求之后，我们编写的YAML的信息就被提交给了APIServer。

而APIServer的第一个功能，就是过滤这个请求，并完成一些前置性的工作，比如授权、超时处理、审计等。

然后，请求会进入MUX和Routes流程。MUX和Routes是APIServer完成URL和Handler绑定的场所。而APIServer的Handler要做的事情，就是按照上述匹配流程，找到对应的CronJob类型定义。

接着，APIServer最重要的职责就来了：根据这个CronJob类型定义，使用用户的YAML文件里的字段，创建一个CronJob对象。

而在这个过程中，APIServer会进行一个Convert工作，即：把用户提交的YAML文件，转换成一个叫作Super Version的对象，它正是该API资源类型所有版本的字段全集。这样用户提交的不同版本的YAML文件，就都可以用这个Super Version对象来进行处理了。

接下来，APIServer会先后进行Admission()和Validation()操作。Admission Controller和Initializer，就都属于Admission的内容。

而Validation，则负责验证验证这个对象里的各个字段是否合法。这个被验证过的API对象，都保存在了APIServer里一个叫作Registry的数据结构中。也就是说，只要一个API对象的定义能在Registry里查到，它就是一个有效的Kubernetes API对象。

最后，APIServer会把验证过的API对象转换成用户最初提交的版本，进行序列化操作，并调用Etcd的API把它保存起来。

由此可见，声明式API对于Kubernetes来说非常重要。所以，APIServer这样一个在其他项目里“平淡无奇”的组件，却成了Kubernetes项目的重中之重。它不仅是Google Borg设计思想的集中体现，也是Kubernetes项目里唯一一个被Google公司和RedHat公司双重控制、其他势力根本无法参与其中的组件。

此外，由于同时要兼顾性能、API完备性、版本化、向后兼容等很多工程化指标，所以Kubernetes团队在APIServer项目里大量使用了Go语言的代码生成功能，来自动化注入Convert、DeepCopy等于API资源相关的操作。这部分自动生成的代码，曾一度占到Kubernetes项目总代码的20%~20%。

这也是为何，在过去很长一段时间里，在这样一个极其“复杂”的APIServer中，添加一个Kubernetes风格的API资源类型，是一个非常困难的工作。

CRD

不过，在Kubernetes v1.7之后，这个工作就变得轻松得多了。这，得益于一个全新的API插件机制：CRD。

CRD的全称是Custom Resource Definition。顾名思义，它指的就是，允许用户在Kubernetes中添加一个跟Pod、Node类似的、新的API资源类型，即：自定义API资源。

举个🌰：现在要为Kubernetes添加一个名叫Network的API资源类型。

它的作用是，一旦用户创建一个Network对象，那么Kubernetes就应该使用这个对象定义的网络参数，调用真实的网络插件，比如Neturon项目，为用户创建一个真正的“网络”。这样，将来用户创建的Pod，就可以声明使用这个“网络”了。

Custom Controller

🚀code：https://github.com/Khighness/highness-network-controller

为Network这个自定义API对象编写一个自定义控制器（Custom Controller）。

“声明式API”并不像”命令式API“那样有着明显的执行逻辑。这就使得基于声明式API的业务功能实现，往往需要控制器模式来“监视”API对象的变化（比如，创建或者删除Network），然后以此来决定实际要执行的具体工作。

总的来说，编写自定义控制器代码的过程包括：编写main函数、编写自定义控制器的定义，以及编写控制器里的业务逻辑三个部分。

Main Function

main函数的主要工作就是，定义并出示好一个自定义控制器（Custom Controller），然后启动它。这部分的主要内容如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func main() {
	flag.Parse()
	// handler the first shutdown signal gracefully
	stopCh := signals.SetupSignalHandler()

	// 1.1 build config via the url of APIServer and the path of kubeConfig
	cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeConfig)
	// 1.2 create the client of kubernetes
	kubeClient, err := kubernetes.NewForConfig(cfg)
	// 1.3 create the client of network
	networkClient, err := clientset.NewForConfig(cfg)

	// 2. create a factory for network, and transfer if to controller
	networkInformerFactory := informers.NewSharedInformerFactory(networkClient, 30*time.Second)
	controller := NewController(kubeClient, networkClient, networkInformerFactory.Samplecrd().V1().Networks())

	// 3. start the factory and the controller
	go networkInformerFactory.Start(stopCh)
	if err = controller.Run(2, stopCh); err != nil {
		glog.Fatalf("Error running controller: %s", err.Error())
	}
}

这个main函数主要通过三步完成了初始化并启动了一个自定义控制器的工作。

Working Principle

一个自定义控制器的工作原理，可以用下面这样一幅流程图来表示：


从这幅示意图的最左边看起。

这个控制器要做的第一件事，是从Kubernetes的APIServer里获取它所关心的对象，也就是定义的Network对象。

这个操作，依靠的是一个叫做Informer（可以翻译为：通知器）的代码库完成的。Informer与API对象是一一对应的，所以传递给自定义控制器的，正是一个Network对象的Informer（Network Informer）。

在创建这个Informer工厂的时候，需要给它传递一个networkClient。

事实上，Network Informer正是使用这个networkClient，跟APIServer建立了连接。不过，真正负责维护这个连接的，则是Informer所使用的Reflector包。

更具体地说，Reflector使用的是一种叫作ListAndWatch的方法，来获取并监听这些Network对象实例的变化。

在ListAndWatch机制下，一旦APIServer端有新的Network实例被创建、删除或者更新，Reflector都会收到事件通知。这时，该事件及它对应的API对象这个组合，就被称为增量（Delta），它会被放进一个Delta FIFO Queue（增量先进先出队列）中。

而另一方面，Informer会不断地从这个Delta FIFO Queue里读去（Pop）增量。每拿到一个增量，Informer就会判断这个增量里的事件类型，然后创建或者更新本地对象的缓存。这个缓存，在Kubernetes里一般被叫作Store。

比如，如果事件类型是Added（添加对象），那么Informer就会通过一个叫作Indexer的库把这个增量里的API对象保存到本地缓存中，并为它创建索引。相反，如果增量的事件类型是Deleted（删除对象），那么Informer就会从本地缓存中删除这个对象。

这个同步缓存的工作，是Informer的第一个职责，也是它最重要的职责。

而Informer的第二个职责，则是根据这些事件的类型，触发事先注册好的ResourceEventHandler。这些Handler，需要在创建控制器的时候注册给它对应的Informer。

接下来，编写这个这个控制器的定义，它的主要内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
func NewController(
	kubeClientSet kubernetes.Interface,
	networkClientSet clientset.Interface,
	networkInformer informers.NetworkInformer) *Controller {

	runtime.Must(networkschema.AddToScheme(scheme.Scheme))
	glog.V(4).Info("Creating event broadcaster")
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClientSet.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

	controller := &Controller{
		kubeClientSet:    kubeClientSet,
		networkClientSet: networkClientSet,
		networkListener:  networkInformer.Lister(),
		networkSynced:    networkInformer.Informer().HasSynced,
		workQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Networks"),
		recorder:         recorder,
	}

	glog.Info("Setting up event handlers")
	networkInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueNetwork,
		UpdateFunc: func(oldObj, newObj interface{}) {
			oldNetwork := oldObj.(*samplecrdv1.Network)
			newNetwork := newObj.(*samplecrdv1.Network)
			if oldNetwork.ResourceVersion == newNetwork.ResourceVersion {
				return
			}
			controller.enqueueNetwork(newObj)
		},
		DeleteFunc: controller.enqueueNetworkForDelete,
	})

	return controller
}

在main函数中创建了两个client（kubeClient和networkClient），然后在这段代码里，使用这两个client和前端创建的Informer，初始化了自定义控制器。

值得注意的是，在这个自定义控制器里，设置了一个工作队列（work queue），它正是处于示意图中间位置的WorkQueue。这个工作队列的作用是，负责同步Informer和控制循环之间的数据。

实际上，Kubernetes提供了很多个工作队列的实现，可以根据需要选择合适的库直接使用。

然后，为networkInformer注册了三个Handler（AddFunc、UpdateFunc和DeleteFunc），分别对应API对象的“添加”“更新”和”删除“事件。而具体的处理操作，都是将该事件对应的API对象加入到工作队列中。

需要注意的是，实际入队的并不是API对象本身，而是它们本身的Key，即：该API对象的namespace/name。

而后面即将编写的控制循环，则会不断地从这个工作队列里拿到这些Key，然后开始执行真正的控制逻辑。

综上所述，所谓Informer，其实就是一个带有本地缓存和索引机制的、可以注册EventHandler的client。它是自定义控制器跟APIServer进行数据同步的重要组件。

更具体地说，Informer通过一种叫作ListAndWatch的方法，把APIServer的API对象缓存在了本地，并负责更新和维护这个缓存。

其中，ListAndWatch方法的含义是：首先，通过APIServer的LIST API“获取”所有最新版本的API对象；然后，再通过WATCH API来“监听”所有这些API对象的变化。

而通过监听到的事件变化，Informer就可以实时更新本地缓存，并且调用这些事件对应的EventHandler了。

此外，在这个过程中，每经过resyncPeroid指定的时间，Informer维护的本地缓存，都会使用最近一次LIST返回的结果强制更新一次，从而保证缓存的有效性。在Kubernetes中，这个缓存强制更新的操作就叫做：resync。

需要注意的是，这个定时resync操作，也会触发Informer注册的“更新”事件。但此时，这个“更新”事件对应的Network对象实际上并没有发生变化，即：新、旧两个Network对象的ResourceVersion是一样的。在这种情况下，Informer就不需要对这个更新事件再做进一步的处理了。

这也是为什么上面的UpdateFunc方法里，先判断了一下新、旧两个Network对象的版本（ResourceVersion）是否发生了变化，然后才开始进行的入队操作。

以上，就是Kubernetes中的Informer库的工作原理。

接下来，我们即来到了示意图中最后面的控制循环（Control Loop）部分，也正是main函数最后调用controller.Run()启动的“控制循环”。 它的主要内容如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
	defer runtime.HandleCrash()
	defer c.workQueue.ShutDown()

	glog.Info("Starting network control loop")

	glog.Info("Waiting for informers caches to sync")
	if ok := cache.WaitForCacheSync(stopCh, c.networkSynced); !ok {
		return fmt.Errorf("failed to wait for caches to sync")
	}

	glog.Info("Starting workers")
	for i := 0; i < threadiness; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	glog.Info("Started workers")
	<-stopCh
	glog.Info("Shutting down workers")

	return nil
}

可以看到，启动控制循环的逻辑非常简单：

  • 首先，等待Infomer完成一次本地缓存的数据同步操作；
  • 然后，直接通过goroutine启动一个（或者并发启动多个）“无限循环”的任务。

而这个“无限循环”任务的每一个循环周期，执行的正是我们真正的业务逻辑。

接下来，就编写这个自定义控制器的业务逻辑，它的主要内容如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
func (c *Controller) runWorker() {
	for c.processNextWorkItem() {
	}
}

func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.workQueue.Get()
	if shutdown {
		return false
	}

	err := func(obj interface{}) error {
		defer c.workQueue.Done(obj)
		var (
			key string
			ok  bool
		)
		if key, ok = obj.(string); !ok {
			c.workQueue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workQueue but get %#v", obj))
			return nil
		}
		if err := c.syncHandler(key); err != nil {
			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
		}
		c.workQueue.Forget(obj)
		glog.Infof("Successfully sync '%s'", key)
		return nil
	}(obj)

	if err != nil {
		runtime.HandleError(err)
		return true
	}

	return true
}

func (c *Controller) syncHandler(key string) error {
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(err)
		return nil
	}

	network, err := c.networkListener.Networks(namespace).Get(name)
	if err != nil {
		if errors.IsNotFound(err) {
			glog.Warningf("Network: %s/%s does not exist in local cache, will delete it from Neutron ...",
				namespace, name)
			glog.Infof("[Neutron] Deleting network: %s/%s ...", namespace, name)
			return nil
		}
		runtime.HandleError(fmt.Errorf("failed to list nerwork by: %s/%s", namespace, name))
		return err
	}

	glog.Infof("[Neutron] Try to process network: %#v ...", network)
	c.recorder.Event(network, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
	return nil
}

可以看到，在这个执行周期里（processNextWorkItem），首先从工作队列里出队（workqueue.Get）了一个成员，也就是一个Key（Network对象的：namespace/name）。

然后，在syncHandler方法中，使用这个Key，尝试从Informer维护的缓存中拿到了它所对应的Network对象。

可以看到，在这里，使用了networkListener来尝试获取这个Key对应的Network对象。这个操作，其实就是在访问本地缓存的索引。实际上，在Kubernetes的源码中，可以经常看到控制器从各种Listener里获取对象，比如：podListener、nodeListener等等，它们使用的都是Informer和缓存机制。

而如果控制循环从缓存中拿不到这个（即：networkListener返回了isNotFound错误），那就意味着这个Network对象的Key是通过前面的“删除”事件添加进工作队列的。所以，尽管队列里有这个Key，但是对应的Network对象已经被删除了。

这时候就需要调用Neutron的API，把这个Key的Neutron网络从真实的集群里删除掉。

而如果能够获取到对应的Network对象，就可与执行控制器模式里的对比“期望状态”和“实际状态”的逻辑了。

其中，自定义控制器“千辛万苦”拿到的这个Network对象，正是APIServer里保存的”期望状态”，即：用户通过YAML文件提交到APIServer里的信息。当然，在例子里，它已经被Informer缓存在了本地。

那么，“实际状态”又从哪里来呢？

当然是来自于实际的集群了，所以，控制循环需要通过Neutron API来查询实际的网络情况。

比如，可以通过Neutron来查询这个Network对象对应的真实网络是否存在。

  • 如果不存在，这就是一个典型的“期望状态”与“实际状态”不一致的情形。这时，就需要使用这个Network对象里的信息（比如：CIDR和Gateway），调用Neutron API来创建真实的网络。
  • 如果存在，那么，就要读取这个真实网络的信息，判断它是否跟Network对象里的信息一致，从而决定是否要通过Neutron来更新这个已经存在的真实网络。

这样，就可以通过对比“期望状态”和“实际状态”的差异，完成了一次调谐（Reconcile）的过程。

Build and Run

可以通过如下流程将项目编译成二进制文件：

1
2
3
4
$ git clone https://github.com/Khighness/highness-network-controller
$ cd highness-network-controller
$ go mod tidy
$ go build -o highness-network-controller .

尝试运行自定义控制器，如下所示：

1
2
3
4
5
6
$ ./highness-network-controller -kubeconfig=$HOME/.kube/config -alsologtostderr=true
I0617 09:21:41.741895    5389 controller.go:77] Setting up event handlers
I0617 09:21:41.746107    5389 controller.go:103] Starting network control loop
I0617 09:21:41.746139    5389 controller.go:106] Waiting for informers caches to sync
E0617 09:21:41.772859    5389 reflector.go:134] pkg/mod/k8s.io/client-go@v0.0.0-20180902073250-f06dbfd73543/tools/cache/reflector.go:95: Failed to list *v1.Network: the server could not find the requested resource (get networks.samplecrd.k8s.io)
E0617 09:21:42.775466    5389 reflector.go:134] pkg/mod/k8s.io/client-go@v0.0.0-20180902073250-f06dbfd73543/tools/cache/reflector.go:95: Failed to list *v1.Network: the server could not find the requested resource (get networks.samplecrd.k8s.io)

可以看到，一开始启动后会报错。

这是因为，此时Network对象的CRD还没有被创建出来，所以Informer去APIServer里“获取”（List）Network对象时，并不能找到Network这个API资源类型的定义。

接下来，创建Network对象的CRD：

1
2
$ kubectl apply -f crd/network.yaml
customresourcedefinition.apiextensions.k8s.io/networks.samplecrd.k8s.io created

这时候，可以看到控制器的日志恢复了正常，控制循环启动成功：

1
2
I0617 09:22:45.946500    5389 controller.go:111] Starting workers
I0617 09:22:45.946518    5389 controller.go:117] Started workers

接下来，创建一个Network对象：

1
2
3
4
5
6
7
8
9
10
11
$ cat example/example-network.yaml
apiVersion: samplecrd.k8s.io/v1
kind: Network
metadata:
  name: example-network
spec:
  cidr: "192.168.0.0/16"
  gateway: "192.168.0.1"

$ kubectl apply -f example/example-network.yaml
network.samplecrd.k8s.io/example-network created

这时候，查看一下控制器的日志：

1
2
3
I0617 09:23:31.766052    5389 controller.go:189] [Neutron] Try to process network: &v1.Network{TypeMeta:v1.TypeMeta{Kind:"", APIVersion:""}, ObjectMeta:v1.ObjectMeta{Name:"example-network", GenerateName:"", Namespace:"default", SelfLink:"/apis/samplecrd.k8s.io/v1/namespaces/default/networks/example-network", UID:"597cfbff-00f0-447c-a8d5-6cec45c26222", ResourceVersion:"23354", Generation:1, CreationTimestamp:time.Date(2022, time.June, 17, 9, 23, 31, 0, time.Local), DeletionTimestamp:<nil>, DeletionGracePeriodSeconds:(*int64)(nil), Labels:map[string]string(nil), Annotations:map[string]string{"kubectl.kubernetes.io/last-applied-configuration":"{\"apiVersion\":\"samplecrd.k8s.io/v1\",\"kind\":\"Network\",\"metadata\":{\"annotations\":{},\"name\":\"example-network\",\"namespace\":\"default\"},\"spec\":{\"cidr\":\"192.168.0.0/16\",\"gateway\":\"192.168.0.1\"}}\n"}, OwnerReferences:[]v1.OwnerReference(nil), Initializers:(*v1.Initializers)(nil), Finalizers:[]string(nil), ClusterName:""}, Spec:v1.NetworkSpec{Cidr:"192.168.0.0/16", Gateway:"192.168.0.1"}} ...
I0617 09:23:31.766972    5389 controller.go:155] Successfully sync 'default/example-network'
I0617 09:23:31.767115    5389 event.go:221] Event(v1.ObjectReference{Kind:"Network", Namespace:"default", Name:"example-network", UID:"597cfbff-00f0-447c-a8d5-6cec45c26222", APIVersion:"samplecrd.k8s.io/v1", ResourceVersion:"23354", FieldPath:""}): type: 'Normal' reason: 'Synced' Network synced successfully

再修改一下YAML文件的内容，并提交更新，如下所示：

1
2
3
4
5
6
7
8
9
10
11
$ cat example/example-network.yaml
apiVersion: samplecrd.k8s.io/v1
kind: Network
metadata:
  name: example-network
spec:
  cidr: "192.168.0.0/16"
  gateway: "192.168.1.1"
  
$ kubectl apply -f example/example-network.yaml
network.samplecrd.k8s.io/example-network configured

这时候，查看一下控制器的日志：

1
2
3
I0617 09:32:57.516092    5389 controller.go:189] [Neutron] Try to process network: &v1.Network{TypeMeta:v1.TypeMeta{Kind:"", APIVersion:""}, ObjectMeta:v1.ObjectMeta{Name:"example-network", GenerateName:"", Namespace:"default", SelfLink:"/apis/samplecrd.k8s.io/v1/namespaces/default/networks/example-network", UID:"597cfbff-00f0-447c-a8d5-6cec45c26222", ResourceVersion:"24722", Generation:4, CreationTimestamp:time.Date(2022, time.June, 17, 9, 23, 31, 0, time.Local), DeletionTimestamp:<nil>, DeletionGracePeriodSeconds:(*int64)(nil), Labels:map[string]string(nil), Annotations:map[string]string{"kubectl.kubernetes.io/last-applied-configuration":"{\"apiVersion\":\"samplecrd.k8s.io/v1\",\"kind\":\"Network\",\"metadata\":{\"annotations\":{},\"name\":\"example-network\",\"namespace\":\"default\"},\"spec\":{\"cidr\":\"192.168.0.0/16\",\"gateway\":\"192.168.1.1\"}}\n"}, OwnerReferences:[]v1.OwnerReference(nil), Initializers:(*v1.Initializers)(nil), Finalizers:[]string(nil), ClusterName:""}, Spec:v1.NetworkSpec{Cidr:"192.168.0.0/16", Gateway:"192.168.1.1"}} ...
I0617 09:32:57.516378    5389 controller.go:155] Successfully sync 'default/example-network'
I0617 09:32:57.516393    5389 event.go:221] Event(v1.ObjectReference{Kind:"Network", Namespace:"default", Name:"example-network", UID:"597cfbff-00f0-447c-a8d5-6cec45c26222", APIVersion:"samplecrd.k8s.io/v1", ResourceVersion:"24722", FieldPath:""}): type: 'Normal' reason: 'Synced' Network synced successfully

最后，删除这个对象：

1
2
$ kubectl delete -f example/example-network.yaml
network.samplecrd.k8s.io "example-network" deleted

这时候，查看一下控制器的日志：

1
2
3
W0617 09:35:56.386759   11958 controller.go:180] Network: default/example-network does not exist in local cache, will delete it from Neutron ...
I0617 09:35:56.386790   11958 controller.go:182] [Neutron] Deleting network: default/example-network ...
I0617 09:35:56.386797   11958 controller.go:155] Successfully sync 'default/example-network'

这一次，在控制器的日志里，可以看到Informer注册的“删除”事件被触发，并且控制循环“调用”Neutron API“删除”了真实环境里的网络。

Expand

实际上，这套流程不仅可以用在自定义API资源上，也完全可以用在Kubernetes原生的默认API对象上。

比如，在main函数里，除了创建一个Network Informer外，还可以初始化一个Kubernetes默认API对象的 Informer工厂，比如Deployment对象的Informer。具体做法如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
func main() {
  ...
  
  kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
  
  controller := NewController(kubeClient, exampleClient,
  kubeInformerFactory.Apps().V1().Deployments(),
  networkInformerFactory.Samplecrd().V1().Networks())
  
  go kubeInformerFactory.Start(stopCh)
  ...
}

在这段代码里，首先使用Kubernetes的client（kubeClient）创建了一个工厂；

然后，用跟Network类似的处理方法，生成了一个Deployment Informer；

接着，把Deployment Informer传递给了自定义控制器；当然，还需要调用Start方法来启动这个Deployment Informer。

而有了这个Deployment Informer后，这个控制器也就持有了所有Deployment对象的信息。接下来，它既可以通过deploymentInformer.Listener()来获取Etcd里所有Deployment对象，也可用为这个Deployment Informer注册具体的Handler。

更重要的是，这就使得在这个自定义控制器里面，可以通过对自定义API对象和默认API对象进行协同，从而实现更加复杂的编排功能。

比如：用户每创建一个新的Deployment，这个自定义控制器，就可以为它创建一个对应的Network供它使用。

Summary

所谓的Informer，就是一个自带缓存和索引机制，可以触发Handler的客户端库。这个本地缓存在Kubernetes中一般被称为Store，索引一般被称为Index。

Informer使用了Reflector包，它是一个可以通过ListAndWatch机制获取并监视API对象变化的客户端封装。

Reflector和Informer之间，用到了一个“增量先进先出队列”进行协同。而Informer与需要编写的控制循环之间，则使用了一个工作队列来进行协同。

在实际应用中，除了控制循环之外的所有代码，实际上都是Kubernetes自动生成的，即：pkg/client/{informers, listers, clientset}里的内容。

而这些自动生成的代码，就为我们提供了一个可靠而高效地获取 API 对象“期望状态”的编程库。

所以，接下来，作为开发者，只需要关注如何拿到“实际状态”，然后如何拿它去跟“期望状态”做对比，从而决定接下来要做的业务逻辑即可。