K8s之CRD
上一节主要讲述了K8s的动态准入控制器Initializers，这一节我们介绍另一种扩展K8sAPI的方式，CRD。
一、CRD概述
Kubernetes中将一切都视为资源，资源是Kubernetes API中的一个Endpoint，存储了某种类型的API对象，例如Pod、Deployment、Configmap、Volume等都是一种资源，不过这些都属于内置资源，是Kubernetes默认提供的资源类型，在Kubernetes V1.7之后提供了一种自定义资源，代表某种自定义的配置或者独立运行的服务，这种自定义资源类型就是CRD（CustomerResourceDefinition）。
CRD作为一种Kubernetes扩展资源存在，就是为了满足Kubernetes自身不具备的一些功能，可以通过CRD保证新的资源快速的注册和使用，要使用CRD，你可以把它理解成两部分，Customer Resource和Customer Controller，Customer Resource可以让用户简单的存储和获取结构化数据，将这种对象动态的注册到Kubernetes集群中。而Customer Controller可以把上面自定义的资源更新成用户想要的状态。
二、CRD的使用场景
1.提供、管理外部数据存储，因为Customer Resource就是一种存在Etcd中的数据结构，你可以将外部数据存储到Etcd中而不使用单独的数据库，用Kubernetes的声明式API去管理他的生命周期。
2.对Kubernetes的基础资源进行更高层次的抽象，例如一些自定义控制器，你既可以管理自定义的资源状态，也可以改变Kubernetes原有资源，如Ingress-controller。
三、Customer Controller工作机制
下面是官方的一张控制器的架构图，其中蓝色部分为client-go提供的框架，红色部分是我们需要自己实现扩展业务的逻辑。
我们先来介绍下蓝色组件
1.Clients：负责与APIServer交互的客户端
2.Informer：监控目标资源的变化
3.Workqueue：暂存资源变更事件的工作队列
我们详细介绍一下Informer，他依赖于Kubernetes的List/Watch API，是一个可监听事件并触发回调函数的二级缓存工具包。Informer主要包含Controller、Reflector、DeltaFiFO、LocalStore、Lister和Processor六个组件，这个Controller和Customer Controller没有任何关系，他的作用是管理Informer的生命周期，记录其缓存信息，controller的数据结构如下：
ReFlector的作用是通过Kubernetes的Watch API监听某个资源下的所有事件，DeltaFIFO和LocalStore是Informer的二级缓存，DeltaFIFO用来存储Reflector Watch的各种事件，而LocalStore存储的是Reflector List的所有事件，Lister用来做List和Get操作，Processor中记录了所有的回调函数，也就是ResourceEventHandler。这个回调函数的触发就会调用上面架构图中红色部分CallBacks。Processor的结构如下：
不明白没关系，我们用一个Deployment的创建来串一下整个Informer的逻辑
在Informer初始化的时候，Reflector组件会调用List API将所有Deployment资源存储到LocalStore中，当你创建一个新的Deployment的时候，Reflector会通过Watch API获取此次事件并将其放到DeltaFIFO队列中，然后查看LocalStore中是否存在这个deployment，如果没有DeltaFIFO会pop这个事件到Controller中，Controller触发Processor的回调函数OnAdd。如果是Get一个已有的Deployment，Lister会直接在LocalStore中Get而不是直接去访问APIServer，从而实现二级缓存的作用。
Workqueue就是处理资源变化事件的队列和重试机制，它包括延迟队列和限速队列，由上面得架构图可以看出来是回调函数ResourceEventHandler调用的Workqueue，ReourceEventHandler会在Workqueue中添加一个以命名空间/资源构成的key，通过queue.add(key)的方式入队后，经过queue.Get()获取第一个Key进行处理，处理成功调用queue.Forget(key)清除key，并且调用queue.Done()彻底删除该事件，处理失败调用queue.AddRateLimited(key)重新入队。
obj, shutdown := c.workqueue.Get()
if shutdown {
return false
}
err := func(obj interface{}) error {
defer c.workqueue.Done(obj)
var key string
var ok bool
if key, ok = obj.(string); !ok {
c.workqueue.Forget(obj)
utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
return nil
}
if err := c.syncHandler(key); err != nil {
c.workqueue.AddRateLimited(key)
return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
}
c.workqueue.Forget(obj)
klog.Infof("Successfully synced '%s'", key)
return nil
控制器会调用Worker去处理workqueue中的事件（即架构图中右侧的红色部分），上面代码中的scyncHandler(key)即为用户自定义的Worker，下面这个官方demo中给出的业务逻辑，它实现了创建一个deploment名为Foo中定义的deploymentName的deployment的功能
func (c *Controller) syncHandler(key string) error {
// Convert the namespace/name string into a distinct namespace and name
namespace, name, err := cache.SplitMetaNamespaceKey(key)
if err != nil {
utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
return nil
}
// Get the Foo resource with this namespace/name
foo, err := c.foosLister.Foos(namespace).Get(name)
if err != nil {
// The Foo resource may no longer exist, in which case we stop
processing.
if errors.IsNotFound(err) {
utilruntime.HandleError(fmt.Errorf("foo '%s' in work queue no longer exists", key))
return nil
}
return err
}
deploymentName := foo.Spec.DeploymentName
if deploymentName == "" {
// We choose to absorb the error here as the worker would requeue the
resource otherwise. Instead, the next time the resource is updated
the resource will be queued again.
utilruntime.HandleError(fmt.Errorf("%s: deployment name must be specified", key))
return nil
}
// Get the deployment with the name specified in Foo.spec
deployment, err := c.deploymentsLister.Deployments(foo.Namespace).Get(deploymentName)
// If the resource doesn't exist, we'll create it
if errors.IsNotFound(err) {
deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Create(newDeployment(foo))
}
// If an error occurs during Get/Create, we'll requeue the item so we can
attempt processing again later. This could have been caused by a
temporary network failure, or any other transient reason.
if err != nil {
return err
}
// If the Deployment is not controlled by this Foo resource, we should log
a warning to the event recorder and ret
if !metav1.IsControlledBy(deployment, foo) {
msg := fmt.Sprintf(MessageResourceExists, deployment.Name)
c.recorder.Event(foo, corev1.EventTypeWarning, ErrResourceExists, msg)
return fmt.Errorf(msg)
}
// If this number of the replicas on the Foo resource is specified, and the
number does not equal the current desired replicas on the Deployment, we
should update the Deployment resource.
if foo.Spec.Replicas != nil && *foo.Spec.Replicas != *deployment.Spec.Replicas {
klog.V(4).Infof("Foo %s replicas: %d, deployment replicas: %d", name, *foo.Spec.Replicas, *deployment.Spec.Replicas)
deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Update(newDeployment(foo))
}
// If an error occurs during Update, we'll requeue the item so we can
attempt processing again later. THis could have been caused by a
temporary network failure, or any other transient reason.
if err != nil {
return err
}
// Finally, we update the status block of the Foo resource to reflect the
current state of the world
err = c.updateFooStatus(foo, deployment)
if err != nil {
return err
}
c.recorder.Event(foo, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
return nil
}
四、CRD的配置和使用
官方提供了一个CRD的demo帮助我们更好地理解CRD
使用go get下载项目https://github.com/kubernetes/sample-controller
go build -o sample-controller .
./sample-controller -kubeconfig=$HOME/.kube/config
2.定义一个资源类型
kubectl create -f crd.yaml
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:name: foos.samplecontroller.k8s.io
spec:group: samplecontroller.k8s.io
version: v1alpha1
names: kind: Foo
plural: foos
scope: Namespaced
3.创建一个资源实例
kubectl create -f example.yaml
apiVersion: samplecontroller.k8s.io/v1alpha1
kind: Foo
metadata:name: example-foo
spec:deploymentName: example-foo
replicas: 1
kubectl get deploy会发现自动创建出来一个example-foo的deployment。
