Client-go入门及实战

kubectl get pods -v=6

GET https://<API-SERVER>/api/v1/namespaces
GET https://<API-SERVER>/api/v1/namespaces?limit=500

GET https://<API-SERVER>/apis/apps/v1/namespaces/default/deployments

apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx

package main

import (
	"context"
	"flag"
	"fmt"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	// 1. 确定 kubeconfig 路径：优先用命令行参数，缺省用 ~/.kube/config
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "kubeconfig 路径")
	} else {
		flag.StringVar(&kubeconfig, "kubeconfig", "", "kubeconfig 路径")
	}
	flag.Parse()

	// 2. 从 kubeconfig 构建 REST 配置
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		panic(err.Error())
	}

	// 3. 用配置初始化 Clientset
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	ctx := context.Background()

	// 4. 列出 default 下的所有 Pod（core 组）
	pods, err := clientset.CoreV1().Pods("default").List(ctx, metav1.ListOptions{})
	if err != nil {
		panic(err.Error())
	}
	fmt.Println("== Pods in default ==")
	for _, p := range pods.Items {
		fmt.Println("  ", p.Name)
	}

	// 5. 列出 default 下的所有 Deployment（apps 组）
	deployments, err := clientset.AppsV1().Deployments("default").List(ctx, metav1.ListOptions{})
	if err != nil {
		panic(err.Error())
	}
	fmt.Println("== Deployments in default ==")
	for _, d := range deployments.Items {
		fmt.Println("  ", d.Name)
	}
}

go mod init client-example
go mod tidy
go run main.go

options := metav1.ListOptions{LabelSelector: "app=nginx"}

config, err := rest.InClusterConfig()
if err != nil {
	panic(err.Error())
}
clientset, err := kubernetes.NewForConfig(config)

/var/run/secrets/kubernetes.io/serviceaccount/
├── ca.crt      # K8s 的 CA 证书
└── token       # 用于请求 API Server 的令牌（JWT）

KUBERNETES_SERVICE_HOST      # 例如 10.96.0.1
KUBERNETES_SERVICE_PORT      # 例如 443

# 阶段一：编译 Go 程序
FROM golang:1.22 AS build
WORKDIR /app
COPY . .
RUN go mod download
RUN CGO_ENABLED=0 go build -o main main.go

# 阶段二：精简运行镜像
FROM alpine:latest
COPY --from=build /app/main /
ENTRYPOINT ["/main"]

docker build -t <registry>/client-demo:outer .
docker push <registry>/client-demo:outer

apiVersion: apps/v1
kind: Deployment
metadata:
  name: client-demo
spec:
  replicas: 1
  selector:
    matchLabels:
      app: client-demo
  template:
    metadata:
      labels:
        app: client-demo
    spec:
      containers:
        - name: app
          image: <registry>/client-demo:outer
          imagePullPolicy: Always

E ... can not list resource "pods": ... is forbidden: User "system:serviceaccount:default:default"

# 创建 Role：允许 list 命名空间 default 下的 pod 和 deployment
kubectl create role client-demo-role --verb=list,get,watch \
  --resource=pods,deployments -n default

# 创建 RoleBinding：把该 Role 绑定到 default 命名空间的 default SA
kubectl create rolebinding client-demo-rb \
  --role=client-demo-role \
  --serviceaccount=default:default -n default

package main

import (
	"context"
	"flag"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "")
	}
	flag.Parse()

	// 从 kubeconfig 构建基础 REST 配置
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		panic(err)
	}

	// 使用 RESTClient 需要额外补齐三样配置
	config.GroupVersion = &schema.GroupVersion{Group: "", Version: "v1"} // 目标：core group v1
	config.APIPath = "/api"                                               // core group 前缀是 /api
	config.NegotiatedSerializer = serializer.WithoutConversionCodecFactory{
		CodecFactory: scheme.Codecs,
	}

	// 用补齐后的 config 初始化 RESTClient
	client, err := rest.RESTClientFor(config)
	if err != nil {
		panic(err)
	}

	// 声明一个空结构体，用于承接反序列化后的 PodList
	var podList corev1.PodList

	// 拼出 GET /api/v1/namespaces/kube-system/pods?limit=500 并解码
	err = client.Get().
		Namespace("kube-system").
		Resource("pods").
		VersionedParams(&metav1.ListOptions{}, scheme.ParameterCodec).
		Do(context.Background()).
		Into(&podList)
	if err != nil {
		panic(err)
	}

	for _, p := range podList.Items {
		fmt.Println("  ", p.Name, " -> ", p.Namespace)
	}
}

package main

import (
	"context"
	"flag"
	"fmt"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "")
	}
	flag.Parse()

	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		panic(err)
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err)
	}

	replicas := int32(1) // replicas 是指针类型，需要先取地址

	// 用结构体描述一个 Deployment —— 等价于写一份 Deployment YAML
	deployment := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name: "nginx",
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: &replicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{"app": "nginx"},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"app": "nginx"},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:  "nginx",
							Image: "nginx:latest",
							Ports: []corev1.ContainerPort{
								{ContainerPort: 80},
							},
						},
					},
				},
			},
		},
	}

	result, err := clientset.AppsV1().
		Deployments("default").
		Create(context.Background(), deployment, metav1.CreateOptions{})
	if err != nil {
		if !apierrors.IsAlreadyExists(err) { // 兼容已存在的情况
			panic(err)
		}
		fmt.Println("deployment already exists")
		return
	}

	fmt.Printf("created deployment %s/%s\n", result.Namespace, result.Name)
}

package main

import (
	"context"
	"flag"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "")
	}
	flag.Parse()

	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		panic(err)
	}
	dynClient, err := dynamic.NewForConfig(config)
	if err != nil {
		panic(err)
	}

	// 声明 GVR：我们要操作 core 组 v1 的 pod
	gvr := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}

	// 注意：返回的不是 corev1.PodList，而是一个非结构化对象
	unstructList, err := dynClient.Resource(gvr).
		Namespace("kube-system").
		List(context.Background(), metav1.ListOptions{})
	if err != nil {
		panic(err)
	}

	// 把非结构化数据转换成标准 PodList
	var podList corev1.PodList
	err = runtime.DefaultUnstructuredConverter.FromUnstructured(
		unstructList.UnstructuredContent(), &podList,
	)
	if err != nil {
		panic(err)
	}

	for _, p := range podList.Items {
		fmt.Println("  ", p.Name)
	}
}

package main

import (
	"context"
	"flag"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "")
	}
	flag.Parse()

	config, _ := clientcmd.BuildConfigFromFlags("", kubeconfig)
	clientset, _ := kubernetes.NewForConfig(config)

	ctx := context.Background()
	// 为方便测试，Watch 60 秒
	watcher, err := clientset.CoreV1().
		Pods("default").
		Watch(ctx, metav1.ListOptions{TimeoutSeconds: int64Ptr(60)})
	if err != nil {
		panic(err)
	}

	// 遍历事件通道
	for event := range watcher.ResultChan() {
		pod, ok := event.Object.(*corev1.Pod) // 断言成 Pod
		if !ok {
			continue
		}
		// 根据事件类型分派到业务逻辑
		switch event.Type {
		case watch.Added:
			processPod("ADDED", pod.Name)
		case watch.Modified:
			processPod("MODIFIED", pod.Name)
		case watch.Deleted:
			processPod("DELETED", pod.Name)
		default:
			fmt.Println("unknown event", event.Type)
		}
	}
	fmt.Println("watch 通道关闭，退出")
}

func processPod(action, name string) {
	fmt.Printf("[%s] %s\n", action, name)
	// 这里可以写你的业务逻辑，比如触发扩容、发告警……
}

func int64Ptr(v int64) *int64 { return &v }

package main

import (
	"context"
	"flag"
	"fmt"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		flag.StringVar(&kubeconfig, "kubeconfig", home+"/.kube/config", "")
	}
	flag.Parse()

	config, _ := clientcmd.BuildConfigFromFlags("", kubeconfig)
	clientset, _ := kubernetes.NewForConfig(config)

	// 新建共享 informer 工厂；第二个参数 12h 是"兜底 resync"周期
	factory := informers.NewSharedInformerFactory(clientset, 12*time.Hour)

	// 用工厂分别创建 Deployment 和 Service 的 informer
	depInformer := factory.Apps().V1().Deployments()
	svcInformer := factory.CoreV1().Services()

	// 给 Deployment 挂事件回调
	depInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			fmt.Println("Deployment ADDED:", obj.(*appsv1.Deployment).Name)
		},
		UpdateFunc: func(old, new interface{}) {
			fmt.Println("Deployment UPDATED:", new.(*appsv1.Deployment).Name)
		},
		DeleteFunc: func(obj interface{}) {
			fmt.Println("Deployment DELETED:", obj.(*appsv1.Deployment).Name)
		},
	})

	// 给 Service 挂事件回调（示例只挂 Add）
	svcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			fmt.Println("Service ADDED:", obj.(*corev1.Service).Name)
		},
	})

	// 接收终止信号，让程序能优雅退出
	stopCh := make(chan struct{})
	defer close(stopCh)

	// 启动所有 informer 的 List & Watch
	factory.Start(stopCh)

	// 等待缓存同步完成（第一次全量列表拉完）
	factory.WaitForCacheSync(stopCh)

	// 从本地缓存直接读取 default 下的所有 Deployment（不走 API 请求）
	deps, _ := depInformer.Lister().Deployments("default").List(labels.Everything())
	fmt.Println("== cached deployments ==")
	for _, d := range deps {
		fmt.Println("  ", d.Name)
	}

	// 阻塞主 goroutine，保持监听
	<-stopCh
}

type controller struct {
	indexer  cache.Indexer                    // 从 informer 拿到的本地缓存索引器
	queue    workqueue.TypedRateLimitingInterface[string] // 限速工作队列
	informer informers.DeploymentInformer    // Deployment informer
}

func newController(
	queue workqueue.TypedRateLimitingInterface[string],
	informer informers.DeploymentInformer,
) *controller {
	return &controller{
		indexer:  informer.Informer().GetIndexer(),
		queue:    queue,
		informer: informer,
	}
}

factory := informers.NewSharedInformerFactory(clientset, 12*time.Hour)
depInformer := factory.Apps().V1().Deployments()

// 限速队列：key 的类型用 string
queue := workqueue.NewTypedRateLimitingQueue[string](
	workqueue.DefaultTypedControllerRateLimiter[string](),
)

// 事件回调统一做一件事：计算 key 并入队
depInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
	AddFunc:    func(obj interface{}) { addDeployment(obj, queue) },
	UpdateFunc: func(old, new interface{}) { updateDeployment(old, new, queue) },
	DeleteFunc: func(obj interface{}) { deleteDeployment(obj, queue) },
})

c := newController(queue, depInformer)

stopCh := make(chan struct{})
factory.Start(stopCh)
factory.WaitForCacheSync(stopCh)

// 开一个 worker goroutine 消费队列
go func() {
	for c.processNextItem() {
	}
}()
<-stopCh

func addDeployment(obj interface{}, queue workqueue.TypedRateLimitingInterface[string]) {
	key, _ := cache.MetaNamespaceKeyFunc(obj) // 生成 "default/nginx" 这样的 key
	queue.Add(key)
}
func updateDeployment(old, new interface{}, queue workqueue.TypedRateLimitingInterface[string]) {
	key, _ := cache.MetaNamespaceKeyFunc(new)
	queue.Add(key)
}
func deleteDeployment(obj interface{}, queue workqueue.TypedRateLimitingInterface[string]) {
	// delete 事件对象可能被包装成 DeletedFinalStateUnknown，处理更健壮
	key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	queue.Add(key)
}

func (c *controller) processNextItem() bool {
	// 从队列取出一个 key；若队列已关闭则返回 false
	key, shutdown := c.queue.Get()
	if shutdown {
		return false
	}
	// 处理完成后必须调用 Done（这决定是否重新入队）
	defer c.queue.Done(key)

	err := c.syncToStdout(key)
	if err == nil {
		// 成功：forget，之后不再重试
		c.queue.Forget(key)
		return true
	}
	// 失败：交给 handleErr 决定是否重试
	c.handleErr(err, key)
	return true
}

func (c *controller) syncToStdout(key string) error {
	obj, exists, err := c.indexer.GetByKey(key)
	if err != nil {
		return err
	}
	if !exists {
		fmt.Printf("object %q not found (可能已被删除)\n", key)
		return nil
	}
	dep, ok := obj.(*appsv1.Deployment)
	if !ok {
		return fmt.Errorf("unexpected type")
	}
	fmt.Printf("[process] %s/%s\n", key, dep.Name)

	// 故意制造一个错误场景：名字等于 test-deployment 即失败，验证重试
	if dep.Name == "test-deployment" {
		return fmt.Errorf("simulate: deployment %s failed", key)
	}
	return nil
}

func (c *controller) handleErr(err error, key string) {
	// 因为失败次数过多而放弃重试（这里是 <5 才继续）
	if c.queue.NumRequeues(key) < 5 {
		fmt.Printf("error processing %q, retrying: %v\n", key, err)
		c.queue.AddRateLimited(key) // 重新入队并限速（指数退避）
		return
	}
	// 重试次数够了，放弃
	c.queue.Forget(key)
	fmt.Printf("drop key %q after too many retries: %v\n", key, err)
}

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: myresources.example.com
spec:
  group: example.com
  names:
    kind: MyResource
    listKind: MyResourceList
    plural: myresources      # 复数形式 → 请求路径/URI 里的 resource
    singular: myresource     # 单数形式
  scope: Namespaced          # 基于命名空间隔离
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                key1: { type: string }   # 两个示例字段
                key2: { type: string }

kubectl apply -f crd.yaml

cat <<EOF | kubectl apply -f -
apiVersion: example.com/v1alpha1
kind: MyResource
metadata:
  name: myresource-sample
spec:
  key1: hello
  key2: world
EOF

package main

import (
	"context"
	"errors"
	"fmt"
	"os"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	// 期望用法： go run main.go get myresource
	if len(os.Args) != 3 {
		panic("usage: go run main.go <get|other> <kind>")
	}
	command, kind := os.Args[1], os.Args[2]
	if command != "get" {
		panic("本示例只实现了 get")
	}

	var kubeconfig string
	if home := homedir.HomeDir(); home != "" {
		kubeconfig = home + "/.kube/config"
	}
	config, _ := clientcmd.BuildConfigFromFlags("", kubeconfig)

	// 1) 需要 Clientset 来拿 Discovery
	clientset, _ := kubernetes.NewForConfig(config)
	// 2) 需要 DynamicClient 来真正取数据
	dynClient, _ := dynamic.NewForConfig(config)

	// 3) 用 DiscoveryClient 枚举服务端支持的资源清单
	dc := clientset.Discovery()
	// ServerPreferredResources 返回服务端偏爱的资源清单 []*meta.APIResourceList
	apiGroupResources, err := dc.ServerPreferredResources()
	if err != nil && !errors.IsNotFound(err) && len(apiGroupResources) == 0 {
		panic(err)
	}

	// 4) 用 RESTMapper 基于这些资源建立 GVK<->GVR 映射
	restMapper := restmapper.NewDiscoveryRESTMapper(apiGroupResources)

	// 5) 构造目标 GVK
	gvk := schema.GroupVersionKind{
		Group:   "example.com",
		Version: "v1alpha1",
		Kind:    kind, // MyResource
	}

	// 6) GVK -> GVR（RESTMapping 的的核心封装）
	mapping, err := restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		panic(err)
	}
	gvr := mapping.Resource // 这才是真正的 GVR

	fmt.Println("resolved GVR:", gvr)

	// 7) 用 DynamicClient 按 GVR 去 List
	list, err := dynClient.Resource(gvr).
		Namespace("default").
		List(context.Background(), metav1.ListOptions{})
	if err != nil {
		panic(err)
	}

	// 8) 打印（非结构化数据）
	for _, item := range list.Items {
		name := item.GetName()
		rv := item.GetResourceVersion()
		fmt.Printf("  %s  (resourceVersion=%s)\n", name, rv)
	}
}

go run main.go get MyResource