首页 资讯 社群 我的社区 搜索

【Kubernetes】深入解析声明式API

LM123
2019-05-28 18:00:07

在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'
...
用户评论