從生成CRD到編寫自定義控制器教程示例
介紹
我們可以使用code-generator 以及controller-tools來進行代碼自動生成，通過代碼自動生成可以幫我們自動生成 CRD 資源對象,以及客戶端訪問的 ClientSet、Informer、Lister 等工具包，接下來我們就來瞭解下如何編寫一個自定義的控制器。
CRD定義
首先初始化項目：
$ mkdir operator-crd && cd operator-crd $ go mod init operator-crd $ mkdir -p pkg/apis/example.com/v1
在該文件夾下新建doc.go文件，內容如下所示：
// +k8s:deepcopy-gen=package // +groupName=example.com package v1
根據 CRD 的規范定義，這裡我們定義的 group 為example.com，版本為v1，在頂部添加瞭一個代碼自動生成的deepcopy-gen的 tag，為整個包中的類型生成深拷貝方法。
然後就是非常重要的資源對象的結構體定義，新建types.go文件，types.go內容可以使用type-scaffpld自動生成,具體文件內容如下：
package v1
import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
// BarSpec defines the desired state of Bar
type BarSpec struct {
// INSERT ADDITIONAL SPEC FIELDS -- desired state of cluster
DeploymentName string `json:"deploymentName"`
Image string `json:"image"`
Replicas *int32 `json:"replicas"`
}
// BarStatus defines the observed state of Bar.
// It should always be reconstructable from the state of the cluster and/or outside world.
type BarStatus struct {
// INSERT ADDITIONAL STATUS FIELDS -- observed state of cluster
}
// 下面這個一定不能少，少瞭的話不能生成 lister 和 informer
// +genclient
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// Bar is the Schema for the bars API
// +k8s:openapi-gen=true
type Bar struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec BarSpec `json:"spec,omitempty"`
Status BarStatus `json:"status,omitempty"`
}
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// BarList contains a list of Bar
type BarList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitempty"`
Items []Bar `json:"items"`
}
然後可以參考系統內置的資源對象，還需要提供 AddToScheme 與 Resource 兩個變量供 client 註冊，新建 register.go 文件，內容如下所示：
package v1
import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
)
// SchemeGroupVersion 註冊自己的自定義資源
var SchemeGroupVersion = schema.GroupVersion{Group: "example.com", Version: "v1"}
// Kind takes an unqualified kind and returns back a Group qualified GroupKind
func Kind(kind string) schema.GroupKind {
return SchemeGroupVersion.WithKind(kind).GroupKind()
}
// Resource takes an unqualified resource and returns a Group qualified GroupResource
func Resource(resource string) schema.GroupResource {
return SchemeGroupVersion.WithResource(resource).GroupResource()
}
var (
// SchemeBuilder initializes a scheme builder
SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
// AddToScheme is a global function that registers this API group & version to a scheme
AddToScheme = SchemeBuilder.AddToScheme
)
// Adds the list of known types to Scheme.
func addKnownTypes(scheme *runtime.Scheme) error {
// 添加 Bar 與 BarList這兩個資源到 scheme
scheme.AddKnownTypes(SchemeGroupVersion,
&Bar{},
&BarList{},
)
metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
return nil
}
使用controller-gen生成crd：
$ controller-gen crd paths=./... output:crd:dir=crd
生成example.com_bars.yaml文件如下所示：
---
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
annotations:
controller-gen.kubebuilder.io/version: (devel)
creationTimestamp: null
name: bars.example.com
spec:
group: example.com
names:
kind: Bar
listKind: BarList
plural: bars
singular: bar
scope: Namespaced
versions:
- name: v1
schema:
openAPIV3Schema:
description: Bar is the Schema for the bars API
properties:
apiVersion:
description: 'APIVersion defines the versioned schema of this representation
of an object. Servers should convert recognized schemas to the latest
internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
type: string
kind:
description: 'Kind is a string value representing the REST resource this
object represents. Servers may infer this from the endpoint the client
submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
type: string
metadata:
type: object
spec:
description: BarSpec defines the desired state of Bar
properties:
deploymentName:
description: INSERT ADDITIONAL SPEC FIELDS -- desired state of cluster
type: string
image:
type: string
replicas:
format: int32
type: integer
required:
- deploymentName
- image
- replicas
type: object
status:
description: BarStatus defines the observed state of Bar. It should always
be reconstructable from the state of the cluster and/or outside world.
type: object
type: object
served: true
storage: true
最終項目結構如下所示：
$ tree
.
├── crd
│ └── example.com_bars.yaml
├── go.mod
├── go.sum
└── pkg
└── apis
└── example.com
└── v1
├── doc.go
├── register.go
└── types.go
5 directories, 6 files
生成客戶端相關代碼
上面我們準備好資源的 API 資源類型後，就可以使用開始生成 CRD 資源的客戶端使用的相關代碼瞭。
首先創建生成代碼的腳本，下面這些腳本均來源於sample-controller提供的示例：
$ mkdir hack && cd hack
在該目錄下面新建 tools.go 文件，添加 code-generator 依賴，因為在沒有代碼使用 code-generator 時，go module 默認不會為我們依賴此包。文件內容如下所示：
// +build tools // 建立 tools.go 來依賴 code-generator // 因為在沒有代碼使用 code-generator 時，go module 默認不會為我們依賴此包. package tools import _ "k8s.io/code-generator"
然後新建 update-codegen.sh 腳本，用來配置代碼生成的腳本：
#!/usr/bin/env bash
set -o errexit
set -o nounset
set -o pipefail
SCRIPT_ROOT=$(dirname "${BASH_SOURCE[0]}")/..
CODEGEN_PKG=${CODEGEN_PKG:-$(cd "${SCRIPT_ROOT}"; ls -d -1 ./vendor/k8s.io/code-generator 2>/dev/null || echo ../code-generator)}
bash "${CODEGEN_PKG}"/generate-groups.sh "deepcopy,client,informer,lister" \
operator-crd/pkg/client operator-crd/pkg/apis example.com:v1 \
--output-base "${SCRIPT_ROOT}"/../ \
--go-header-file "${SCRIPT_ROOT}"/hack/boilerplate.go.txt
# To use your own boilerplate text append:
# --go-header-file "${SCRIPT_ROOT}"/hack/custom-boilerplate.go.txt
同樣還有 verify-codegen.sh 腳本，用來校驗生成的代碼是否是最新的：
#!/usr/bin/env bash
set -o errexit
set -o nounset
set -o pipefail
SCRIPT_ROOT=$(dirname "${BASH_SOURCE[0]}")/..
DIFFROOT="${SCRIPT_ROOT}/pkg"
TMP_DIFFROOT="${SCRIPT_ROOT}/_tmp/pkg"
_tmp="${SCRIPT_ROOT}/_tmp"
cleanup() {
rm -rf "${_tmp}"
}
trap "cleanup" EXIT SIGINT
cleanup
mkdir -p "${TMP_DIFFROOT}"
cp -a "${DIFFROOT}"/* "${TMP_DIFFROOT}"
"${SCRIPT_ROOT}/hack/update-codegen.sh"
echo "diffing ${DIFFROOT} against freshly generated codegen"
ret=0
diff -Naupr "${DIFFROOT}" "${TMP_DIFFROOT}" || ret=$?
cp -a "${TMP_DIFFROOT}"/* "${DIFFROOT}"
if [[ $ret -eq 0 ]]
then
echo "${DIFFROOT} up to date."
else
echo "${DIFFROOT} is out of date. Please run hack/update-codegen.sh"
exit 1
fi
還有一個為生成的代碼文件添加頭部內容的 boilerplate.go.txt 文件，內容如下所示，其實就是為每個生成的代碼文件頭部添加上下面的開源協議聲明：
/*
Copyright The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
接下來我們就可以來執行代碼生成的腳本瞭，首先將依賴包放置到 vendor 目錄中去：
$ go mod vendor
然後執行腳本生成代碼：
$ chmod +x ./hack/update-codegen.sh $./hack/update-codegen.sh Generating deepcopy funcs Generating clientset for example.com:v1 at operator-crd/pkg/client/clientset Generating listers for example.com:v1 at operator-crd/pkg/client/listers Generating informers for example.com:v1 at operator-crd/pkg/client/informers
代碼生成後，整個項目的 pkg 包變成瞭下面的樣子：
$ tree pkg
pkg
├── apis
│ └── example.com
│ └── 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
│ └── example.com
│ └── v1
│ ├── bar.go
│ ├── doc.go
│ ├── example.com_client.go
│ ├── fake
│ │ ├── doc.go
│ │ ├── fake_bar.go
│ │ └── fake_example.com_client.go
│ └── generated_expansion.go
├── informers
│ └── externalversions
│ ├── example.com
│ │ ├── interface.go
│ │ └── v1
│ │ ├── bar.go
│ │ └── interface.go
│ ├── factory.go
│ ├── generic.go
│ └── internalinterfaces
│ └── factory_interfaces.go
└── listers
└── example.com
└── v1
├── bar.go
└── expansion_generated.go
20 directories, 26 files
仔細觀察可以發現pkg/apis/example.com/v1目錄下面多瞭一個zz_generated.deepcopy.go文件,在pkg/client文件夾下生成瞭 clientset和 informers 和 listers 三個目錄，有瞭這幾個自動生成的客戶端相關操作包，我們就可以去訪問 CRD 資源瞭，可以和使用內置的資源對象一樣去對 Bar 進行 List 和 Watch 操作瞭。
編寫控制器
首先要先獲取訪問資源對象的 ClientSet，在項目根目錄下面新建 main.go 文件。
package main
import (
"k8s.io/client-go/tools/clientcmd"
"k8s.io/klog/v2"
clientset "operator-crd/pkg/client/clientset/versioned"
"operator-crd/pkg/client/informers/externalversions"
"time"
"os"
"os/signal"
"syscall"
)
var (
onlyOneSignalHandler = make(chan struct{})
shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}
)
// 註冊 SIGTERM 和 SIGINT 信號
// 返回一個 stop channel， 該通道在捕獲到第一個信號時被關閉
// 如果捕獲到第二個信號，程序直接退出
func setupSignalHandler() (stopCh <-chan struct{}) {
// 當調用兩次的時候 panics
close(onlyOneSignalHandler)
stop := make(chan struct{})
c := make(chan os.Signal, 2)
// Notify 函數讓 signal 包將輸入信號轉發到c
// 如果沒有列出要傳遞的信號，會將所有輸入信號傳遞到 c； 否則隻會傳遞列出的輸入信號
signal.Notify(c, shutdownSignals...)
go func() {
<-c
close(stop)
<-c
os.Exit(1) // 第二個信號直接退出
}()
return stop
}
func main() {
stopCh := setupSignalHandler()
// 獲取config
config, err := clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
if err != nil {
klog.Fatalln(err)
}
// 通過config構建clientSet
// 這裡的clientSet 是 Bar 的
clientSet, err := clientset.NewForConfig(config)
if err != nil {
klog.Fatalln(err)
}
// informerFactory 工廠類， 這裡註入我們通過代碼生成的 client
// client 主要用於和 API Server 進行通信，實現 ListAndWatch
factory := externalversions.NewSharedInformerFactory(clientSet, time.Second*30)
// 實例化自定義控制器
controller := NewController(factory.Example().V1().Bars())
// 啟動 informer，開始list 和 watch
go factory.Start(stopCh)
// 啟動控制器
if err = controller.Run(2, stopCh); err != nil {
klog.Fatalf("Error running controller: %s", err.Error())
}
}
首先初始化一個用於訪問 Bar 資源的 ClientSet 對象，然後同樣新建一個 Bar 的 InformerFactory 實例，通過這個工廠實例可以去啟動 Informer 開始對 Bar 的 List 和 Watch 操作，然後同樣我們要自己去封裝一個自定義的控制器，在這個控制器裡面去實現一個控制循環，不斷對 Bar 的狀態進行調諧。
在項目根目錄下新建controller.go文件，內容如下所示：
package main
import (
"fmt"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"
v1 "operator-crd/pkg/apis/example.com/v1"
"time"
informers "operator-crd/pkg/client/informers/externalversions/example.com/v1"
)
type Controller struct {
informer informers.BarInformer
workqueue workqueue.RateLimitingInterface
}
func NewController(informer informers.BarInformer) *Controller {
controller := &Controller{
informer: informer,
// WorkQueue 的實現，負責同步 Informer 和控制循環之間的數據
workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "bar"),
}
klog.Info("Setting up Bar event handlers")
// informer 註冊瞭三個 Handler（AddFunc、UpdateFunc 和 DeleteFunc）
// 分別對應 API 對象的“添加”“更新”和“刪除”事件。
// 而具體的處理操作，都是將該事件對應的 API 對象加入到工作隊列中
informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: controller.addBar,
UpdateFunc: controller.updateBar,
DeleteFunc: controller.deleteBar,
})
return controller
}
func (c *Controller) Run(thread int, stopCh <-chan struct{}) error {
defer runtime.HandleCrash()
defer c.workqueue.ShuttingDown()
// 記錄開始日志
klog.Info("Starting Bar control loop")
klog.Info("Waiting for informer caches to sync")
// 等待緩存同步數據
if ok := cache.WaitForCacheSync(stopCh, c.informer.Informer().HasSynced); !ok {
return fmt.Errorf("failed to wati for caches to sync")
}
klog.Info("Starting workers")
for i := 0; i < thread; i++ {
go wait.Until(c.runWorker, time.Second, stopCh)
}
klog.Info("Started workers")
<-stopCh
klog.Info("Shutting down workers")
return nil
}
// runWorker 是一個不斷運行的方法，並且一直會調用 c.processNextWorkItem 從 workqueue讀取消息
func (c *Controller) runWorker() {
for c.processNExtWorkItem() {
}
}
// 從workqueue讀取和讀取消息
func (c *Controller) processNExtWorkItem() bool {
// 獲取 item
item, shutdown := c.workqueue.Get()
if shutdown {
return false
}
if err := func(item interface{}) error {
// 標記以及處理
defer c.workqueue.Done(item)
var key string
var ok bool
if key, ok = item.(string); !ok {
// 判讀key的類型不是字符串，則直接丟棄
c.workqueue.Forget(item)
runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", item))
return nil
}
if err := c.syncHandler(key); err != nil {
return fmt.Errorf("error syncing '%s':%s", item, err.Error())
}
c.workqueue.Forget(item)
return nil
}(item); err != nil {
runtime.HandleError(err)
return false
}
return true
}
// 嘗試從 Informer 維護的緩存中拿到瞭它所對應的 Bar 對象
func (c *Controller) syncHandler(key string) error {
namespace, name, err := cache.SplitMetaNamespaceKey(key)
if err != nil {
runtime.HandleError(fmt.Errorf("invalid respirce key:%s", key))
return err
}
bar, err := c.informer.Lister().Bars(namespace).Get(name)
if err != nil {
if errors.IsNotFound(err) {
// 說明是在刪除事件中添加進來的
return nil
}
runtime.HandleError(fmt.Errorf("failed to get bar by: %s/%s", namespace, name))
return err
}
fmt.Printf("[BarCRD] try to process bar:%#v ...", bar)
// 可以根據bar來做其他的事。
// todo
return nil
}
func (c *Controller) addBar(item interface{}) {
var key string
var err error
if key, err = cache.MetaNamespaceKeyFunc(item); err != nil {
runtime.HandleError(err)
return
}
c.workqueue.AddRateLimited(key)
}
func (c *Controller) deleteBar(item interface{}) {
var key string
var err error
if key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(item); err != nil {
runtime.HandleError(err)
return
}
fmt.Println("delete crd")
c.workqueue.AddRateLimited(key)
}
func (c *Controller) updateBar(old, new interface{}) {
oldItem := old.(*v1.Bar)
newItem := new.(*v1.Bar)
// 比較兩個資源版本，如果相同，則不處理
if oldItem.ResourceVersion == newItem.ResourceVersion {
return
}
c.workqueue.AddRateLimited(new)
}
我們這裡自定義的控制器隻封裝瞭一個 Informer 和一個限速隊列，我們當然也可以在裡面添加一個用於訪問本地緩存的 Indexer，但實際上 Informer 中已經包含瞭 Lister，對於 List 和 Get 操作都會去通過 Indexer 從本地緩存中獲取數據，所以隻用一個 Informer 也是完全可行的。
同樣在 Informer 中註冊瞭3個事件處理器，將監聽的事件獲取到後送入 workqueue 隊列，然後通過控制器的控制循環不斷從隊列中消費數據，根據獲取的 key 來獲取數據判斷對象是需要刪除還是需要進行其他業務處理，這裡我們同樣也隻是打印出瞭對應的操作日志，對於實際的項目則進行相應的業務邏輯處理即可。
到這裡一個完整的自定義 API 對象和它所對應的自定義控制器就編寫完畢瞭。
測試
接下來我們直接運行我們的main函數：
I0512 16:51:33.922138 39032 controller.go:29] Setting up Bar event handlers I0512 16:51:33.922255 39032 controller.go:47] Starting Bar control loop I0512 16:51:33.922258 39032 controller.go:48] Waiting for informer caches to sync I0512 16:51:34.023108 39032 controller.go:55] Starting workers I0512 16:51:34.023153 39032 controller.go:60] Started workers
現在我們創建一個Bar資源對象：
# bar.yaml apiVersion: example.com/v1 kind: Bar metadata: name: bar-demo namespace: default spec: image: "nginx:1.17.1" deploymentName: example-bar replicas: 2
直接創建上面的對象，註意觀察控制器的日志：
I0512 16:51:33.922138 39032 controller.go:29] Setting up Bar event handlers I0512 16:51:33.922255 39032 controller.go:47] Starting Bar control loop I0512 16:51:33.922258 39032 controller.go:48] Waiting for informer caches to sync I0512 16:51:34.023108 39032 controller.go:55] Starting workers I0512 16:51:34.023153 39032 controller.go:60] Started workers [BarCRD] try to process bar:"bar-demo" ...
可以看到，我們上面創建 bar.yaml 的操作，觸發瞭 EventHandler 的添加事件，從而被放進瞭工作隊列。然後控制器的控制循環從隊列裡拿到這個對象，並且打印出瞭正在處理這個 bar 對象的日志信息。
同樣我們刪除這個資源的時候，也會有對應的提示。
這就是開發自定義 CRD 控制器的基本流程，當然我們還可以在事件處理的業務邏輯中去記錄一些 Events 信息，這樣我們就可以通過 Event 去瞭解我們資源的狀態瞭，更多關於CRD生成自定義控制器的資料請關註WalkonNet其它相關文章！
推薦閱讀：
- go語言代碼生成器code generator使用示例介紹
- 自動生成代碼controller tool的簡單使用
- 詳解如何讓Go語言中的反射加快
- Golang語言學習拿捏Go反射示例教程
- Go 語言選擇器實例教程