kubernetes client-go

简介

Kubernetes官方从2016年8月份开始，将Kubernetes资源操作相关的核心源码抽取出来，独立出来一个项目Client-go，作为官方提供的Go client。Kubernetes的部分代码也是基于这个client实现的，所以对这个client的质量、性能等方面还是非常有信心的。

client-go是一个调用kubernetes集群资源对象API的客户端，即通过client-go实现对kubernetes集群中资源对象（包括deployment、service、ingress、replicaSet、pod、namespace、node等）的增删改查等操作。大部分对kubernetes进行前置API封装的二次开发都通过client-go这个第三方包来实现

主要package

• kubernetes
  访问 Kubernetes API的一系列的clientset
• discovery
  通过Kubernetes API 进行服务发现
• dynamic
  对任意Kubernetes对象执行通用操作的动态client
• plugin/pkg/client/auth
  可选的身份验证插件，用于从外部来源获取凭据。
• transport
  启动连接和鉴权auth
• tools/cache
  controllers控制器

关键组件

• Reflector
  reflector可以通过ListAndWatch方法对指定的Kubernetes资源进行监听，资源可以是Kubernetes内置的资源类型，也可以是CRD。当一个reflector收到新资源创建的通知时，会通知相应的list API获取对应资源模型并将其写入到一个Delta FIFO的工作队列中
• ClientSet
  提供了直接从Kubernetes API Server 中获取Object的机制，当然使用Lister从本地Cache中获取Objects的方法优先级会高于直接从API Server读取，以最大程度上减轻API server的负载
• Indexer(线程安全)
  indexer的作用是对目标object生成相应的存储索引，在内部实现上，indexer可以支持多种索引方式，同时利用一个线程安全的本地存储持久化对应的object和索引键，默认情况下indexer会使用/的组合形式作为目标object的索引键
• Lister
  从本地Index中获取Objecrts的工作机制，可以在client-go中的tools/cache包中找到其具体实现
• Informer
  Client-go包中一个相对较为高端的设计在于Informer的设计，我们知道我们可以直接通过Kubernetes API交互，但是考虑一点就是交互的形式，Informer设计为List/Watch的方式。Informer在初始化的时先通过List去从Kubernetes API中取出资源的全部object对象，并同时缓存，然后后面通过Watch的机制去监控资源，这样的话，通过Informer及其缓存，我们就可以直接和Informer交互而不是每次都和Kubernetes API交互。
  Informer另外一块内容在于提供了事件handler机制，并会触发回调，这样上层应用如Controller就可以基于回调处理具体业务逻辑。因为Informer通过List、Watch机制可以监控到所有资源的所有事件，因此只要给Informer添加ResourceEventHandler 实例的回调函数实例取实现

  OnAdd(obj interface{}) OnUpdate(oldObj, newObj interface{}) 和 OnDelete(obj interface{})
  

  这三个方法，就可以处理好资源的创建、更新和删除操作
  Kubernetes中都是各种controller的实现，各种controller都会用到Informer

Demo

集群内

package main

import (
    "context"
    "fmt"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "time"
)

func main() {
    // creates the in-cluster config
    config, err := rest.InClusterConfig()
    if err != nil {
        panic(err.Error())
    }
    // creates the clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err.Error())
    }
    for {
        // get pods in all the namespaces by omitting namespace
        // Or specify namespace to get pods in particular namespace
        pods, err := clientset.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{})
        if err != nil {
            panic(err.Error())
        }
        fmt.Printf("There are %d pods in the cluster\n", len(pods.Items))

        // Examples for error handling:
        // - Use helper functions e.g. errors.IsNotFound()
        // - And/or cast to StatusError and use its properties like e.g. ErrStatus.Message
        _, err = clientset.CoreV1().Pods("default").Get(context.TODO(), "example-xxxxx", metav1.GetOptions{})
        if errors.IsNotFound(err) {
            fmt.Printf("Pod example-xxxxx not found in default namespace\n")
        } else if statusError, isStatus := err.(*errors.StatusError); isStatus {
            fmt.Printf("Error getting pod %v\n", statusError.ErrStatus.Message)
        } else if err != nil {
            panic(err.Error())
        } else {
            fmt.Printf("Found example-xxxxx pod in default namespace\n")
        }

        time.Sleep(10 * time.Second)
    }
}

集群外

package main

import (
    "context"
    "flag"
    "fmt"
    "k8s.io/apimachinery/pkg/api/errors"
    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"
    "path/filepath"
    "time"
)

func main() {
    var kubeconfig *string
    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
    }
    flag.Parse()

    // use the current context in kubeconfig
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        panic(err.Error())
    }

    // create the clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err.Error())
    }
    for {
        pods, err := clientset.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{})
        if err != nil {
            panic(err.Error())
        }
        fmt.Printf("There are %d pods in the cluster\n", len(pods.Items))

        // Examples for error handling:
        // - Use helper functions like e.g. errors.IsNotFound()
        // - And/or cast to StatusError and use its properties like e.g. ErrStatus.Message
        namespace := "default"
        pod := "example-xxxxx"
        _, err = clientset.CoreV1().Pods(namespace).Get(context.TODO(), pod, metav1.GetOptions{})
        if errors.IsNotFound(err) {
            fmt.Printf("Pod %s in namespace %s not found\n", pod, namespace)
        } else if statusError, isStatus := err.(*errors.StatusError); isStatus {
            fmt.Printf("Error getting pod %s in namespace %s: %v\n",
                pod, namespace, statusError.ErrStatus.Message)
        } else if err != nil {
            panic(err.Error())
        } else {
            fmt.Printf("Found pod %s in namespace %s\n", pod, namespace)
        }
        time.Sleep(10 * time.Second)
    }

}