前言
最近在极客时间订阅了kubernetes的专栏，这篇文章是想记录一下自己学习CRD(custom resource definition)的过程，加深一下记忆。
准备工作
首先安装一下我们用的go依赖：
cd $GOPATH/src/
mkdir resouer
git clone https://github.com/resouer/k8s-controller-custom-resource.git
cd k8s-controller-custom-resource
godep restore
然后参照github地址，我自己也照着抄了一个，这是我的那个git地址，没有注释，加深一下印象，以下代码还有目录都是我自己创建的，创建目录和文件如下(当前目录)：
hongzhi:k8s-crd hongzhi.wang$ tree
.
├── controller.go
├── crd
│ └── network.yaml
├── example
│ └── example-network.yaml
├── main.go
└── pkg
└── apis
└── samplecrd
├── register.go
└── v1
├── doc.go
├── register.go
└── types.go
并且参考原文填写这些文件，接着安装
go get -u k8s.io/code-generator
cd $GOPATH/src/k8s.io/code-generator
godep restore
然后生成代码
ROOT_PACKAGE="k8s-crd"
CUSTOM_RESOURCE_NAME="samplecrd"
CUSTOM_RESOURCE_VERSION="v1"
cd $GOPATH/src/k8s.io/code-generator
./generate-groups.sh all "$ROOT_PACKAGE/pkg/client" "$ROOT_PACKAGE/pkg/apis" "$CUSTOM_RESOURCE_NAME:$CUSTOM_RESOURCE_VERSION"
脚本运行结果：
hongzhi:code-generator hongzhi.wang$ ./generate-groups.sh all "$ROOT_PACKAGE/pkg/client" "$ROOT_PACKAGE/pkg/apis" "$CUSTOM_RESOURCE_NAME:$CUSTOM_RESOURCE_VERSION"
Generating deepcopy funcs
Generating clientset for samplecrd:v1 at k8s-crd/pkg/client/clientset
Generating listers for samplecrd:v1 at k8s-crd/pkg/client/listers
Generating informers for samplecrd:v1 at k8s-crd/pkg/client/informers
之后代码如下:(自己的types.go:37行的代码注释写错了，然后在生成代码之后，goland 只有 networklist 提示networklist 不是 runtime.Object 类型，而 network 在生成代码之前也提示了，生成之后错误提示就消失了。我仔细看了一下 runtime.Object 是一个 interface 类型，在 k8s.io/apimachinery/pkg/runtime 的目录下的 interface.go 文件中，goland 报错应该是没生成代码之前我们自定义的 type 没有完全实现 runtime.Object 这个接口。)
hongzhi:k8s-crd hongzhi.wang$ tree
.
├── controller.go
├── crd
│ └── network.yaml
├── example
│ └── example-network.yaml
├── main.go
└── pkg
├── apis
│ └── samplecrd
│ ├── register.go
│ └── v1
│ ├── doc.go
│ ├── register.go
│ ├── types.go
│ └── zz_generated.deepcopy.go
├── client
│ ├── clientset
│ │ └── versioned
│ │ ├── clientset.go
│ │ ├── doc.go
│ │ ├── fake
│ │ │ ├── clientset_generated.go
│ │ │ ├── doc.go
│ │ │ └── register.go
│ │ ├── scheme
│ │ │ ├── doc.go
│ │ │ └── register.go
│ │ └── typed
│ │ └── samplecrd
│ │ └── v1
│ │ ├── doc.go
│ │ ├── fake
│ │ │ ├── doc.go
│ │ │ ├── fake_network.go
│ │ │ └── fake_samplecrd_client.go
│ │ ├── generated_expansion.go
│ │ ├── network.go
│ │ └── samplecrd_client.go
│ ├── informers
│ │ └── externalversions
│ │ ├── factory.go
│ │ ├── generic.go
│ │ ├── internalinterfaces
│ │ │ └── factory_interfaces.go
│ │ └── samplecrd
│ │ ├── interface.go
│ │ └── v1
│ │ ├── interface.go
│ │ └── network.go
│ └── listers
│ └── samplecrd
│ └── v1
│ ├── expansion_generated.go
│ └── network.go
└── signals
├── signal.go
├── signal_posix.go
└── signal_windows.go
23 directories, 31 files
signals 目录和里面的内容需要自己创建。接下来创建crd和crd对象
cd $GOPATH/src/k8s-crd
hongzhi:k8s-crd hongzhi.wang$ kubectl apply -f crd/network.yaml
customresourcedefinition.apiextensions.k8s.io/networks.samplecrd.k8s.io created
hongzhi:k8s-crd hongzhi.wang$ kubectl apply -f example/example-network.yaml
network.samplecrd.k8s.io/example-network created
hongzhi:k8s-crd hongzhi.wang$ kubectl get crd
NAME CREATED AT
networks.samplecrd.k8s.io 2018-10-22T07:17:56Z
hongzhi:k8s-crd hongzhi.wang$ kubectl get network
NAME AGE
example-network 18s
整个 cunstom controller 的流程图（张磊在AS深圳2018分享ppt中的图片）:
接下来分析一下 main.go 和 controller.go ：
main.go 中首先需要 kube apiserver 的配置，连上我们k8s集群，然后就是主要的几行代码
networkInformerFactory := informers.NewSharedInformerFactory(networkClient, time.Second*30)
controller := NewController( kubeClient, networkClient,
networkInformerFactory.Smaplecrd().V1().Networks())
go networkInformerFactory.Start(stopCh)
if err = controller.Run(2, stopCh); err != nil{
glog.Fatalf("Error running controller: %s", err.Error())
}
informers.NewSharedInformerFactory 这个函数返回的是SharedInformerFactory这个interface，
然后controller := NewController( kubeClient, networkClient,networkInformerFactory.Smaplecrd().V1().Networks())，这里面主要是把针对这个 crd的增，删，改注册到 eventhandler 中。
这里需要说下，这个增删改是针对 reflector 从 apiserver 接到事件的时候发出的，time.Second*30这个时间是 local cache 和 apiserver 的定时同步时间，同步完成会触发 updateFunc ，这个 update 操作是对所有的network对象的，这时需要对比一下r esource version 如果一样就不需要进入 workqueue。这个增删改正常只是 apiserver 接到请求后传给 reflector 的，比如我们通过 kubectl 删了一个 network 对象，这个删除的操作会通过 apiserver 然后触发 DeleteFunc。
然后调用 SharedInformerFactory 这个 interface 的 Start 方法，其实这个只是启动 reflector 监听 apiserver 中n etwork 对象的变化。
然后调用 func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) 方法启动 controller 。
for i:=0 ; i < threadiness; i++ {
go wait.Until(c.runWorker, time.Second, stopCh)
}
这是要启动 threadiness 个 go 的协程，来进行 reconcile loop。其中期望状态是从 cache 中获取的，实际状态是从 k8s 集群是实时获取的。整个流程是 apiserver 接到请求（增删改），然后发给 reflector ，reflector 调用对应的方法，把这个 key（这个 key 就是 namespace/name 的字符串）加到 workqueue 中，然后根据对应的请求操作 cache ，reconcile loop 从 workqueue 拿到这个 key ，然后从 cache 中取这个 key 如果报 notfound error，reconcile loop 接着就会从实际的集群中删除这个 network 对象。如果 cache 中有这个 key ，那么就看实际状态有没有，没有就创建，有就对比，不一样就更新。
更新于20201015
resync 不是 informer 和 apiserver 去定期通讯，而是 informer 同步最近一次 relist 的结果。之后搞懂了这块会专门写一篇关于 informer resync 的文章。
参考代码
main.go
package main
import (
clientset "k8s-crd/pkg/client/clientset/versioned"
informers "k8s-crd/pkg/client/informers/externalversions"
"flag"
"k8s-crd/pkg/signals"
"k8s.io/client-go/tools/clientcmd"
"github.com/golang/glog"
"k8s.io/client-go/kubernetes"
"time"
)
var (
masterURL string
kubeconfig string
)
func main() {
flag.Parse()
stopCh := signals.SetupSignalHandler()
cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
if err != nil {
glog.Fatalf("Error building kubecofnig: %s", err.Error())
}
kubeClient, err := kubernetes.NewForConfig(cfg)
if err != nil {
glog.Fatalf("Error building kubernetes clientset: %s", err.Error())
}
networkClient, err := clientset.NewForConfig(cfg)
if err != nil {
glog.Fatalf("Error building example clientset: %s", err.Error())
}
networkInformerFactory := informers.NewSharedInformerFactory(networkClient, time.Second*30)
controller := NewController( kubeClient, networkClient,
networkInformerFactory.Smaplecrd().V1().Networks())
go networkInformerFactory.Start(stopCh)
if err = controller.Run(2, stopCh); err != nil{
glog.Fatalf("Error running controller: %s", err.Error())
}
}
func init() {
flag.StringVar(&kubeconfig, "kubeconfig","","Path to kubeconfig")
flag.StringVar(&masterURL, "master", "","The address of the kubernetes api server")
}
controller.go
package main
import (
corev1 "k8s.io/api/core/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
samplecrdv1 "k8s-crd/pkg/apis/samplecrd/v1"
clientset "k8s-crd/pkg/client/clientset/versioned"
networkscheme "k8s-crd/pkg/client/clientset/versioned/scheme"
informers "k8s-crd/pkg/client/informers/externalversions/samplecrd/v1"
listers "k8s-crd/pkg/client/listers/samplecrd/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/workqueue"
"k8s.io/client-go/tools/record"
"k8s.io/client-go/kubernetes/scheme"
"github.com/golang/glog"
"github.com/contrib/service-loadbalancer/Godeps/_workspace/src/k8s.io/kubernetes/pkg/util/runtime"
"fmt"
"k8s.io/apimachinery/pkg/util/wait"
"time"
"k8s.io/apimachinery/pkg/api/errors"
)
const controllerAgentName = "network-controller"
const (
SuccessSynced = "Synced"
MessageResourceSynced = "Network synced successfully"
)
type Controller struct {
kubeclientset kubernetes.Interface
networkclientset clientset.Interface
networksLister listers.NetworkLister
networksSynced cache.InformerSynced
workqueue workqueue.RateLimitingInterface
recorder record.EventRecorder
}
func NewController(
kubeclientset kubernetes.Interface,
networkclientset clientset.Interface,
networkInformer informers.NetworkInformer) *Controller {
utilruntime.Must(networkscheme.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,
networksLister: networkInformer.Lister(),
networksSynced: 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(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
}
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 informer caches to sync")
if ok := cache.WaitForCacheSync(stopCh, c.networksSynced); !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
}
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
var ok bool
if key,ok = obj.(string); !ok {
c.workqueue.Forget(obj)
runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#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 synced '%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(fmt.Errorf("invalid resource key: %s", key))
return nil
}
network, err := c.networksLister.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 network by : %s/%s", namespace, name))
return nil
}
glog.Infof("[Neutron] Try to process network: %#v ...", network)
c.recorder.Event(network, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
return nil
}
func (c *Controller) enqueueNetwork(obj interface{}) {
var key string
var err error
if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil{
runtime.HandleError(err)
return
}
c.workqueue.AddRateLimited(key)
}
func (c *Controller) enqueueNetworkForDelete(obj interface{}) {
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
if err != nil {
runtime.HandleError(err)
return
}
c.workqueue.AddRateLimited(key)
}