如何使用 client-go 監聽自定義資源

在 Kubernetes 中使用 client-go 監聽自定義資源（Custom Resource，簡稱 CR）需要借助 Dynamic Client 或 Custom Informer，因為 client-go 的標準 Clientset 只支持內置資源（如 Pod、Deployment）。自定義資源由 CustomResourceDefinition（CRD）或 Operator 定義，監聽它們需要動態處理其 Group、Version 和 Resource（GVR）。以下是詳細步驟和實現方法。

---

前提條件

• CRD 已部署：確保你的自定義資源定義（CRD）已在集群中注冊。
  • 示例：myresource.example.com/v1，Kind 為 MyResource。
• 依賴：
  • client-go（推薦與集群版本匹配，例如 v0.28.0 對應 Kubernetes 1.28）。
  • 添加依賴：

    go get k8s.io/client-go@v0.28.0
    

• 權限：確保 ServiceAccount 有權訪問 CRD（通過 RBAC 配置）。

---

方法 1：使用 Dynamic Client 和 Informer

Dynamic Client 是 client-go 提供的通用客戶端，支持任意資源類型。結合 SharedInformer，可以監聽自定義資源。

步驟

1. 初始化 Dynamic Client：

   package mainimport ("context""log""time""k8s.io/apimachinery/pkg/apis/meta/v1/unstructured""k8s.io/apimachinery/pkg/runtime/schema""k8s.io/client-go/dynamic""k8s.io/client-go/rest""k8s.io/client-go/tools/cache""k8s.io/client-go/tools/clientcmd"
   )func getDynamicClient() dynamic.Interface {config, err := rest.InClusterConfig()if err != nil {config, err = clientcmd.BuildConfigFromFlags("", "/path/to/kubeconfig")if err != nil {log.Fatalf("Failed to create config: %v", err)}}client, err := dynamic.NewForConfig(config)if err != nil {log.Fatalf("Failed to create dynamic client: %v", err)}return client
   }
   

2. 定義 GVR：

   • 指定自定義資源的 Group、Version 和 Resource。
   • 示例：myresource.example.com/v1，資源名為 myresources。

     gvr := schema.GroupVersionResource{Group:    "example.com",Version:  "v1",Resource: "myresources",
     }
     

3. 創建 Dynamic Informer：

   func main() {client := getDynamicClient()// 創建 Dynamic Informer Factoryfactory := dynamicinformer.NewDynamicSharedInformerFactory(client, time.Minute*30)informer := factory.ForResource(gvr).Informer()// 添加事件處理函數informer.AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc: func(obj interface{}) {unstructuredObj := obj.(*unstructured.Unstructured)name := unstructuredObj.GetName()namespace := unstructuredObj.GetNamespace()log.Printf("CR Added: %s/%s", namespace, name)},UpdateFunc: func(oldObj, newObj interface{}) {oldUnstructured := oldObj.(*unstructured.Unstructured)newUnstructured := newObj.(*unstructured.Unstructured)log.Printf("CR Updated: %s/%s", newUnstructured.GetNamespace(), newUnstructured.GetName())},DeleteFunc: func(obj interface{}) {unstructuredObj := obj.(*unstructured.Unstructured)log.Printf("CR Deleted: %s/%s", unstructuredObj.GetNamespace(), unstructuredObj.GetName())},})// 啟動 Informerctx, cancel := context.WithCancel(context.Background())defer cancel()factory.Start(ctx.Done())factory.WaitForCacheSync(ctx.Done())// 保持運行<-ctx.Done()
   }
   

4. 獲取緩存數據：

   lister := factory.ForResource(gvr).Lister()
   items, err := lister.List(labels.Everything())
   if err != nil {log.Printf("Failed to list CRs: %v", err)
   } else {for _, item := range items {unstructuredObj := item.(*unstructured.Unstructured)log.Printf("Current CR: %s/%s", unstructuredObj.GetNamespace(), unstructuredObj.GetName())}
   }
   

說明

• GVR：通過 kubectl api-resources 查看自定義資源的準確 GVR。
• Unstructured：自定義資源以 unstructured.Unstructured 類型返回，需手動解析字段。
• 依賴：需要導入 k8s.io/client-go/dynamic/informer。

---

方法 2：生成類型化客戶端和 Informer（推薦生產環境）

如果你的 CRD 有明確的 Go 類型（通過代碼生成器生成），可以使用類型化的客戶端和 Informer。這種方法需要更多前期工作，但更安全和直觀。

步驟

1. 生成代碼：

   • 使用 controller-tools 或 k8s.io/code-generator 生成 CRD 的客戶端代碼。
   • 示例 CRD 文件（myresource_v1.yaml）：

     apiVersion: apiextensions.k8s.io/v1
     kind: CustomResourceDefinition
     metadata:name: myresources.example.com
     spec:group: example.comnames:kind: MyResourceplural: myresourcesscope: Namespacedversions:- name: v1served: truestorage: trueschema:openAPIV3Schema:type: objectproperties:spec:type: objectproperties:replicas:type: integer
     

   • 生成命令：

     mkdir -p pkg/apis/example.com/v1
     controller-gen crd paths=./pkg/apis/example.com/v1 output:crd:dir=./manifests
     controller-gen object paths=./pkg/apis/example.com/v1
     k8s.io/code-generator/generate-groups.sh all ./pkg/client ./pkg/apis example.com:v1
     

2. 注冊類型：

   • 在 pkg/apis/example.com/v1/types.go 中定義類型：

     package v1import (metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
     )type MyResourceSpec struct {Replicas int32 `json:"replicas"`
     }type MyResource struct {metav1.TypeMeta   `json:",inline"`metav1.ObjectMeta `json:"metadata,omitempty"`Spec              MyResourceSpec `json:"spec,omitempty"`
     }type MyResourceList struct {metav1.TypeMeta `json:",inline"`metav1.ListMeta `json:"metadata,omitempty"`Items           []MyResource `json:"items"`
     }
     

3. 創建 Informer：

   package mainimport ("context""log""time""k8s.io/client-go/tools/cache"examplev1 "your/module/pkg/apis/example.com/v1"exampleclientset "your/module/pkg/client/clientset/versioned"exampleinformers "your/module/pkg/client/informers/externalversions"
   )func main() {config, err := rest.InClusterConfig()if err != nil {log.Fatalf("Failed to create config: %v", err)}client, err := exampleclientset.NewForConfig(config)if err != nil {log.Fatalf("Failed to create clientset: %v", err)}factory := exampleinformers.NewSharedInformerFactory(client, time.Minute*30)informer := factory.Example().V1().MyResources().Informer()informer.AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc: func(obj interface{}) {myResource := obj.(*examplev1.MyResource)log.Printf("MyResource Added: %s/%s, Replicas: %d", myResource.Namespace, myResource.Name, myResource.Spec.Replicas)},UpdateFunc: func(oldObj, newObj interface{}) {newResource := newObj.(*examplev1.MyResource)log.Printf("MyResource Updated: %s/%s", newResource.Namespace, newResource.Name)},DeleteFunc: func(obj interface{}) {myResource := obj.(*examplev1.MyResource)log.Printf("MyResource Deleted: %s/%s", myResource.Namespace, myResource.Name)},})ctx, cancel := context.WithCancel(context.Background())defer cancel()factory.Start(ctx.Done())factory.WaitForCacheSync(ctx.Done())<-ctx.Done()
   }
   

說明

• 類型安全：使用生成的類型（如 *examplev1.MyResource），避免手動解析。
• 依賴：需要自定義的客戶端包（pkg/client）。
• 復雜度：前期生成代碼較繁瑣，但長期維護更方便。

---

注意事項

1. CRD 注冊：
   • 確保 CRD 已應用（kubectl apply -f myresource_v1.yaml）。
   • 檢查：

     kubectl get crd myresources.example.com
     

2. 權限：
   • 為 ServiceAccount 配置 RBAC：

     apiVersion: rbac.authorization.k8s.io/v1
     kind: ClusterRole
     metadata:name: myresource-reader
     rules:
     - apiGroups: ["example.com"]resources: ["myresources"]verbs: ["get", "list", "watch"]
     ---
     apiVersion: rbac.authorization.k8s.io/v1
     kind: ClusterRoleBinding
     metadata:name: myresource-reader-binding
     subjects:
     - kind: ServiceAccountname: defaultnamespace: default
     roleRef:kind: ClusterRolename: myresource-readerapiGroup: rbac.authorization.k8s.io
     

3. 性能：
   • 使用 WithNamespace 或標簽過濾減少監聽范圍。
4. 錯誤處理：
   • Watch 失敗時，Informer 會自動重試，需關注日志。

---

驗證

1. 創建自定義資源：

   apiVersion: example.com/v1
   kind: MyResource
   metadata:name: test-resourcenamespace: default
   spec:replicas: 3
   

   kubectl apply -f test-resource.yaml
   

2. 運行程序，觀察日志輸出：

   MyResource Added: default/test-resource, Replicas: 3
   

---

總結

• Dynamic Client：
  • 適合快速實現，無需生成代碼。
  • 使用 unstructured.Unstructured 處理數據。
• 類型化客戶端：
  • 適合生產環境，類型安全，需生成代碼。
  • 使用特定類型（如 *MyResource）操作。
• 選擇建議：
  • 測試或簡單場景：Dynamic Client。
  • 長期項目或 Operator：類型化客戶端。

如果你有具體的 CRD 定義或需求（例如監聽特定字段），可以告訴我，我會進一步定制代碼！

---