【Kubernetes】深入解析声明式API

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

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

　　如果现在要声明一个CronJob对象，那么YAML的开始部分会这么写

apiVersion: batch/v2alpha1
kind: CronJob
...

　　CronJob就是这个API对象的资源类型，Batch就是它们的组，v2alpha1就是它的版本

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

 1、Kubernetes会匹配API对象的组

　　对于Kubernetes里的核心API对象（如Pod，Node）是不需要Group的，Kubernetes会直接在/api这个层级进行下一步的匹配过程

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

　　API Group的分类是以对象功能为依据的

2、Kubernetes会进一步匹配到API对象的版本号

　　在Kubernetes中，同一种API对象可以有多个版本，对于会影响到用户的变更就可以通过升级新版本来处理，从而保证了向后兼容。

3、Kubernetes会匹配API对象的资源类型

APIServer创建对象

　　在前面匹配到正确的版本之后，Kubernetes就知道要创建的是一个/apis/batch/v2alpha1下的CronJob对象，APIServer会继续创建这个Cronjob对象。创建过程如下图

1. 当发起创建CronJob的POST请求之后，YAML的信息就被提交给了APIServer，APIServer的第一个功能就是过滤这个请求，并完成一些前置性的工作，比如授权、超时处理、审计等
2. 请求进入MUX和Routes流程，MUX和Routes是APIServer完成URL和Handler绑定的场所。APIServer的Handler要做的事情，就是按照上面介绍的匹配过程，找到对应的CronJob类型定义。
3. 根据这个CronJob类型定义，使用用户提交的YAML文件里的字段，创建一个CronJob对象。这个过程中，APIServer会把用户提交的YAML文件，转换成一个叫做Super Version的对象，它正是该API资源类型所有版本的字段全集，这样用户提交的不同版本的YAML文件，就都可以用这个SuperVersion对象来进行处理了。
4. APIServer会先后进行Admission（如Admission Controller 和 Initializer）和Validation操作（负责验证这个对象里的各个字段是否何方，被验证过得API对象都保存在APIServer里一个叫做Registry的数据结构中）。
5. APIServer会把验证过得API对象转换成用户最初提交的版本，进行系列化操作，并调用Etcd的API把它保存起来。  

 API插件CRD（Custom Resource Definition） 

 　　CRD允许用户在Kubernetes中添加一个跟Pod、Node类似的、新的API资源类型，即：自定义API资源

　　举个栗子，添加一个叫Network的API资源类型，它的作用是一旦用户创建一个Network对象，那么Kubernetes就可以使用这个对象定义的网络参数，调用真实的网络插件，为用户创建一个真正的网络

　　Network对象YAML文件，名叫example-network.yaml,API资源类型是Network，API组是samplecrd.k8s.io，版本是v1

apiVersion: samplecrd.k8s.io/v1
kind: Network
metadata:
  name: example-network
spec:
  cidr: "192.168.0.0/16"
  gateway: "192.168.0.1"

　　上面这个YAML文件，就是一个具体的自定义API资源实例，也叫CR（Custom Resource），为了让Kubernetes认识这个CR，就需要让Kubernetes明白这个CR的宏观定义CRD。接下来编写一个CRD的YAML文件，名字是network.yaml

apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
  name: networks.samplecrd.k8s.io
spec:
  group: samplecrd.k8s.io
  version: v1
  names:
    kind: Network
    plural: networks
  scope: Namespaced

 　　在这个CRD中，指定了“group：samplecrd.k8s.io“ ”version:v1”的API信息，也指定了这个CR的资源类型叫做Network， 复数（plural）是networks。scope是Namespaced，定义的这个Network是属于一个Namespace的对象，类似于Pod

　下面是一些代码部分的操作

　　首先，在GOPATH下创建一个结构如下的项目

$ tree $GOPATH/src/github.com/<your-name>/k8s-controller-custom-resource
.
├── controller.go
├── crd
│   └── network.yaml
├── example
│   └── example-network.yaml
├── main.go
└── pkg
    └── apis
        └── samplecrd
            ├── register.go
            └── v1
                ├── doc.go
                ├── register.go
                └── types.go

　　其中，pkg/apis/samplecrd就是API组的名字，v1是版本，v1下面的types.go文件里，有Network对象的完整描述

　　

　　然后在pkg/api/samplecrd目录下创建了一个register.go文件，用来防止后面要用到的全局变量

package samplecrd

const (
 GroupName = "samplecrd.k8s.io"
 Version   = "v1"
)

　　

　　接着在pkg/api/samplecrd目录下添加一个doc.go文件（Golang的文档源文件）

// +k8s:deepcopy-gen=package

// +groupName=samplecrd.k8s.io
package v1

　　+ <tag_name>[=value]格式的注释，是Kubernetes进行代码生成要用的Annotation风格的注释

　　+k8s:deepcopy-gen=package的意思是请为整个v1包里的所有类型定义自动生成Deepcopy方法

　　+groupName=samplecrd.k8s.io 定义了这个包对应的API组的名称

　　这些定义在doc.go文件的注释，起到的是全局的代码生成控制的作用，所以也被称为Global Tags

　　接下来添加types.go文件，它定义一个Network类型到底有哪些字段（比如，spec字段里的内容）

package v1
...
// +genclient
// +genclient:noStatus
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// Network describes a Network resource
type Network struct {
 // TypeMeta is the metadata for the resource, like kind and apiversion
 metav1.TypeMeta `json:",inline"`
 // ObjectMeta contains the metadata for the particular object, including
 // things like...
 //  - name
 //  - namespace
 //  - self link
 //  - labels
 //  - ... etc ...
 metav1.ObjectMeta `json:"metadata,omitempty"`
 
 Spec networkspec `json:"spec"`
}
// networkspec is the spec for a Network resource
type networkspec struct {
 Cidr    string `json:"cidr"`
 Gateway string `json:"gateway"`
}

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

// NetworkList is a list of Network resources
type NetworkList struct {
 metav1.TypeMeta `json:",inline"`
 metav1.ListMeta `json:"metadata"`
 
 Items []Network `json:"items"`
}

　　+genclient 指请为下面这个API资源类型生成Client代码

　　+genclient:noStatus 指在这个API资源类型定义里，没有Status字段，否则生成的Client就会自动带上UpdateStatus

　　Network类型定义方法和标准的Kubernetes对象一样，都包括的TypeMeta(API 元数据)和OgjectMeta(对象元数据)字段

　　而其中的Spec字段是需要自己定义的部分，在networkspec里，定义个Cidr和Gateway两个字段，其中，每个字段最后面的部分，比如 json:"cidr"，指的就是这个字段被转换成JOSN格式之后的名字，也就是YAML文件里字段的名称

　　除了定义一个Network类型，还需要定义一个Network类型用来描述一组Network对象应该包括哪些字段。因为在Kubernetes中，获取所以X对象的List()方法，返回值都是List类型，而不是X类型的数组。+genclient只需要写在Network上，而不用写在NetworkList上，因为NetworkList只是一个返回值类型，Network才是主类型。

　　+k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object 指请在生成DeepCopy时，实现Kubernetes提供的runtime.Object接口

　　最后还需要编写一个pkg/apis/samplecrd/v1/register.go文件

　　registry的作用就是注册一个类型给APIServer。其中Network资源类型待在服务器端的注册工作，APIServer会自动完成。但要让客户端知道这个Network资源类型的定义，就需要我们在项目里添加一个register.go文件。

package v1
...
// addKnownTypes adds our types to the API scheme by registering
// Network and NetworkList
func addKnownTypes(scheme *runtime.Scheme) error {
 scheme.AddKnownTypes(
  SchemeGroupVersion,
  &Network{},
  &NetworkList{},
 )
 
 // register the type in the scheme
 metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 return nil
}

　　接着使用Kubernetes提供的代码生成工具（k8s.io/code-generator），为上面定义的Network资源类型自动生成clientset,、informer和lister。其中client就是操作Network对象所需要使用的客户端

# 代码生成的工作目录，也就是我们的项目路径
$ ROOT_PACKAGE="github.com/resouer/k8s-controller-custom-resource"
# API Group
$ CUSTOM_RESOURCE_NAME="samplecrd"
# API Version
$ CUSTOM_RESOURCE_VERSION="v1"

# 安装 k8s.io/code-generator
$ go get -u k8s.io/code-generator/...
$ cd $GOPATH/src/k8s.io/code-generator

# 执行代码自动生成，其中 pkg/client 是生成目标目录，pkg/apis 是类型定义目录
$ ./generate-groups.sh all "$ROOT_PACKAGE/pkg/client" "$ROOT_PACKAGE/pkg/apis" "$CUSTOM_RESOURCE_NAME:$CUSTOM_RESOURCE_VERSION"


$ tree
.
├── controller.go
├── crd
│   └── network.yaml
├── example
│   └── example-network.yaml
├── main.go
└── pkg
    ├── apis
    │   └── samplecrd
    │       ├── constants.go
    │       └── v1
    │           ├── doc.go
    │           ├── register.go
    │           ├── types.go
    │           └── zz_generated.deepcopy.go
    └── client
        ├── clientset
        ├── informers
        └── listers

　　其中pkg/apis/samplecre/v1下面的zz_generated.deepcoy.go文件，就是自动是哪个从的DeepCopy代码文件。整个Client目录都是Kubernetes为Network类型生成的客户端库 。

　下面是在Kubernetes集群里创建一个API对象

　　首先使用network.yaml文件，在Kubernetes中创建Network对象的CRD（Custom Resource Definition）：

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

$ kubectl get crd
NAME                        CREATED AT
networks.samplecrd.k8s.io   2018-09-15T10:57:12Z

　　然后可以创建一个Network对象

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

$ kubectl get network
NAME              AGE
example-network   8s

$ kubectl describe network example-network
Name:         example-network
Namespace:    default
Labels:       <none>
...API Version:  samplecrd.k8s.io/v1
Kind:         Network
Metadata:
  ...
  Generation:          1
  Resource Version:    468239
  ...
Spec:
  Cidr:     192.168.0.0/16
  Gateway:  192.168.0.1

　　

自定义控制器

 　　上面举了一个在Kubernetes里添加API资源的过程，下面将为Network这个自定义API对象编写一个自定义控制器（Custom Controller） 

　　 基于声明式API业务功能的实现，往往需要通过控制器模式来监视API对象的变化（如，创建或删除Network），然后以此来决定实际要执行的具体工作。

　　自定义控制器工作原理

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

　　这个操作依靠的Informer的代码块完成的，Informer与API对象是一一对应的，因此传递给自定义控制器的是一个Network对象的Informer

　　在创建Informer工厂时，需要给它传递一个networkClient，Network Informer使用这个networkClient跟APIServer建立了连接。负责维护这个连接的是Informer所使用的Reflector包中的ListAndWatch方法，它用来获取并监听这些Network对象实例的变化。

　　在ListAndWatch机制下，一旦APIServer端有新的Network实例被创建、删除或者更新，Reflector都会收到事件通知，这时，该事件及它对应的API对象这个组合，就被称为增量Delta，它会被放假一个Delta FIFO Queue中。另一方面，Inform会不断地从这个Delta FIFO Queue读取增量，每拿到一个增量，Informer就会判断这个增量里的事件类型，然后创建或更新本地对象的缓冲。

　　同步本地缓存是Informer的第一个职责，也是它最重要的职责。它的第二个职责是根据这些事件的类型，触发实现注册好的ResourceEventHandler。这些Handler需要在创建控制器的时候注册给它对应的Informer。

       编写main函数

　　定义并初始化一个自定义控制器，然后期待它

func main() {
  ...
  //根据Master配置（APIServer的地址端口和kubeconfig的路径）创建一个Kubernetes的client(kubeClient)和Network对象的client（networkClient）
  cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
  ...
  kubeClient, err := kubernetes.NewForConfig(cfg)
  ...
  networkClient, err := clientset.NewForConfig(cfg)
  ...
  //为Network对象创建一个叫做InformerFactory的工厂，并使用它生产一个Network对象的Informer，传递给控制器
  networkInformerFactory := informers.NewSharedInformerFactory(networkClient, ...)
  
  controller := NewController(kubeClient, networkClient,
  networkInformerFactory.Samplecrd().V1().Networks())
  
  //启动上述Informer
  go networkInformerFactory.Start(stopCh)

  //执行controller.run，启动自定义控制器
  if err = controller.Run(2, stopCh); err != nil {
    glog.Fatalf("Error running controller: %s", err.Error())
  }
}

 　　

　　编写控制器

func NewController(
  kubeclientset kubernetes.Interface,
  networkclientset clientset.Interface,
  networkInformer informers.NetworkInformer) *Controller {
  ...
  controller := &Controller{
    kubeclientset:    kubeclientset,
    networkclientset: networkclientset,
    networksLister:   networkInformer.Lister(),
    networksSynced:   networkInformer.Informer().HasSynced,
    workqueue:        workqueue.NewNamedRateLimitingQueue(...,  "Networks"),
    ...
  }
    networkInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: controller.enqueueNetwork,
    UpdateFunc: func(old, new interface{}) {
      oldNetwork := old.(*samplecrdv1.Network)
      newNetwork := new.(*samplecrdv1.Network)
      if oldNetwork.ResourceVersion == newNetwork.ResourceVersion {
        return
      }
      controller.enqueueNetwork(new)
    },
    DeleteFunc: controller.enqueueNetworkForDelete,
 return controller
}

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

　　并且还设置了一个工作队列（work queue ）,它的作用是复制同步Informer和控制循环之间的数据。

　　然后，为networkInformer注册了三个Handler（AddFunc、 UpdateFunc 和 DeleteFunc），分别对应API对象的添加、更新、删除操作。

　　具体的处理操作，是将该事件对应的API对象加入到工作队列中，实际入队的是API对象的Key，即该API对象的<namespace>/<name>， 控制循环会不断从这个工作队列里拿到这些key，然后开始执行真正的控制逻辑。

　　因而，所谓的Informer，其实就是一个带有本地缓存和索引机制的、可以注册EventHandler的client。它是自定义控制跟API Server进行数据同步的重要组件。Informer通过ListAndWatch方法，把APIServer中的API对象缓存在本地，并负责更新和维护这个缓存。其中ListAndWatch方法的含义是：首先通过APIServer的LIST API获取所有最新版本的API对象，然后再通过WATCH API监听所有这些API对象的变化，通过监听到的事件变化，Informer就可以实时地更新本地缓存，并且调用这些事件对应的EventHandler。在这个过程中，每经过resyncPeriod指定的时间，Informer维护的本地缓存，都会使用最近的LIST返回的结果强制更新一次，从而保证缓存的有效性。在Kubernetes中，这个缓存强制更新的操作叫做：resync 

　　编写控制循环　

func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
 ...
  if ok := cache.WaitForCacheSync(stopCh, c.networksSynced); !ok {
    //等待Informer完成一次本地缓存数据的同步操作
    return fmt.Errorf("failed to wait for caches to sync")
  }
  
  ...
  for i := 0; i < threadiness; i++ {
    // 通过gorouting并发启动多个无限循环的任务
    go wait.Until(c.runWorker, time.Second, stopCh)
  }
  
  ...
  return nil
}

　　

　　编写业务逻辑

func (c *Controller) runWorker() {
  for c.processNextWorkItem() {
  }
}

func (c *Controller) processNextWorkItem() bool {
  obj, shutdown := c.workqueue.Get()
  
  ...
  
  err := func(obj interface{}) error {
    ...
    if err := c.syncHandler(key); err != nil {
     return fmt.Errorf("error syncing '%s': %s", key, err.Error())
    }
    
    c.workqueue.Forget(obj)
    ...
    return nil
  }(obj)
  
  ...
  
  return true
}

func (c *Controller) syncHandler(key string) error {

  namespace, name, err := cache.SplitMetaNamespaceKey(key)
  ...
  
  network, err := c.networksLister.Networks(namespace).Get(name)
  if err != nil {
    if errors.IsNotFound(err) {
      glog.Warningf("Network does not exist in local cache: %s/%s, will delete it from Neutron ...",
      namespace, name)
      
      glog.Warningf("Network: %s/%s does not exist in local cache, will delete it from Neutron ...",
    namespace, name)
    
     // FIX ME: call Neutron API to delete this network by name.
     //
     // neutron.Delete(namespace, name)
     
     return nil
  }
    ...
    
    return err
  }
  
  glog.Infof("[Neutron] Try to process network: %#v ...", network)
  
  // FIX ME: Do diff().
  //
  // actualNetwork, exists := neutron.Get(namespace, name)
  //
  // if !exists {
  //   neutron.Create(namespace, name)
  // } else if !reflect.DeepEqual(actualNetwork, network) {
  //   neutron.Update(namespace, name)
  // }
  
  return nil
}

编译部署

# Clone repo
$ git clone https://github.com/resouer/k8s-controller-custom-resource$ cd k8s-controller-custom-resource

### Skip this part if you don't want to build
# Install dependency
$ go get github.com/tools/godep
$ godep restore
# Build
$ go build -o samplecrd-controller .

$ ./samplecrd-controller -kubeconfig=$HOME/.kube/config -alsologtostderr=true
I0915 12:50:29.051349   27159 controller.go:84] Setting up event handlers
I0915 12:50:29.051615   27159 controller.go:113] Starting Network control loop
I0915 12:50:29.051630   27159 controller.go:116] Waiting for informer caches to sync
E0915 12:50:29.066745   27159 reflector.go:134] github.com/resouer/k8s-controller-custom-resource/pkg/client/informers/externalversions/factory.go:117: Failed to list *v1.Network: the server could not find the requested resource (get networks.samplecrd.k8s.io)
...


$ kubectl apply -f crd/network.yaml
...
I0915 12:50:29.051630   27159 controller.go:116] Waiting for informer caches to sync
...
I0915 12:52:54.346854   25245 controller.go:121] Starting workers
I0915 12:52:54.346914   25245 controller.go:127] Started workers


$ 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

...
I0915 12:50:29.051349   27159 controller.go:84] Setting up event handlers
I0915 12:50:29.051615   27159 controller.go:113] Starting Network control loop
I0915 12:50:29.051630   27159 controller.go:116] Waiting for informer caches to sync
...
I0915 12:52:54.346854   25245 controller.go:121] Starting workers
I0915 12:52:54.346914   25245 controller.go:127] Started workers
I0915 12:53:18.064409   25245 controller.go:229] [Neutron] Try to process network: &v1.Network{TypeMeta:v1.TypeMeta{Kind:"", APIVersion:""}, ObjectMeta:v1.ObjectMeta{Name:"example-network", GenerateName:"", Namespace:"default", ... ResourceVersion:"479015", ... Spec:v1.NetworkSpec{Cidr:"192.168.0.0/16", Gateway:"192.168.0.1"}} ...
I0915 12:53:18.064650   25245 controller.go:183] Successfully synced 'default/example-network'
...