1. 前言
转载请说明原文出处, 尊重他人劳动成果!
源码位置: https://github.com/nicktming/client-go/tree/tming-v13.0/tools/cache
分支: tming-v13.0 (基于v13.0版本)
2. 概括
client-go中提供了三种client可以去访问api-server中的资源.
clientset: 提供集群外部的服务用来访问, 只能访问集群中已有的资源, 对于自定义资源crd无法访问. (除非自己编写相关方法)
dynamic: 提供集群外部的服务用来访问, 可以访问第三方资源crd, 无须再添加方法.
In-cluster: 用于集群里面的pod访问集群资源.
不过这三种
client都依赖rest.
3. clientset
3.1 例子
import (
"flag"
"fmt"
"os"
"path/filepath"
"time"
"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"
)
func main() {
var kubeconfig *string
if home := 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()
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
if err != nil {
panic(err.Error())
}
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err.Error())
}
for {
pods, err := clientset.CoreV1().Pods("").List(metav1.ListOptions{})
if err != nil {
panic(err.Error())
}
fmt.Printf("There are %d pods in the cluster\n", len(pods.Items))
namespace := "default"
pod := "example-xxxxx"
_, err = clientset.CoreV1().Pods(namespace).Get(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)
}
}
func homeDir() string {
if h := os.Getenv("HOME"); h != "" {
return h
}
return os.Getenv("USERPROFILE") // windows
}
3.2 分析
接口与实现类
type Interface interface {
Discovery() discovery.DiscoveryInterface
...
CoreV1() corev1.CoreV1Interface
...
}
该
Interface接口定义了获得k8s中所有资源的方法, 比如获得CoreV1中资源的可以调用该接口的CoreV1()方法.
type Clientset struct {
*discovery.DiscoveryClient
...
coreV1 *corev1.CoreV1Client
...
}
Clientset结构体是Interface的实现类. 既然是实现类, 那需要实现该Interface中的每一个方法, 接下来看看它的CoreV1方法是如何实现的.
CoreV1方法
func (c *Clientset) CoreV1() corev1.CoreV1Interface {
return c.coreV1
}
可以看到该实现方法非常简单, 直接返回该
Clientset对象的coreV1属性. 那就说明该属性在哪里初始化过了, 接下来看一下Clientset的生成方法, 有三种方法:
1.
New(c rest.Interface)传入的是一个可以操作api-server的rest接口实现类, 前面说过, 三种client本质上都是依赖rest接口去操作api-server的.
2.NewForConfigOrDie(c *rest.Config)传入的是一个可以生成rest接口实现类的配置, 根据该配置自己生成一个rest接口的实现类. 但是中途生成rest client过程中出现错误, 程序直接panic退出.
3.NewForConfig(c *rest.Config)与NewForConfigOrDie基本一样, 只是出现错误时返回错误并不会退出程序.
以
NewForConfig为例:
func NewForConfig(c *rest.Config) (*Clientset, error) {
configShallowCopy := *c
if configShallowCopy.RateLimiter == nil && configShallowCopy.QPS > 0 {
if configShallowCopy.Burst <= 0 {
return nil, fmt.Errorf("Burst is required to be greater than 0 when RateLimiter is not set and QPS is set to greater than 0")
}
configShallowCopy.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(configShallowCopy.QPS, configShallowCopy.Burst)
}
var cs Clientset
var err error
...
cs.coreV1, err = corev1.NewForConfig(&configShallowCopy)
if err != nil {
return nil, err
}
...
return &cs, nil
}
可以看到调用
corev1.NewForConfig生成cs.coreV1, 如下所示:
// kubernetes/typed/core/v1/core_client.go
func NewForConfig(c *rest.Config) (*CoreV1Client, error) {
config := *c
if err := setConfigDefaults(&config); err != nil {
return nil, err
}
client, err := rest.RESTClientFor(&config)
if err != nil {
return nil, err
}
return &CoreV1Client{client}, nil
}
根据
config配置生成一个与api-server打交道的rest client, 进而由CoreV1Client包装起来. 现在很清晰了cs.corev1 = &CoreV1Client{client}.
CoreV1Client
接下来看一下
CoreV1Client的结构是如何的.
// kubernetes/typed/core/v1/core_client.go
type CoreV1Interface interface {
RESTClient() rest.Interface
ComponentStatusesGetter
ConfigMapsGetter
EndpointsGetter
EventsGetter
LimitRangesGetter
NamespacesGetter
NodesGetter
PersistentVolumesGetter
PersistentVolumeClaimsGetter
PodsGetter
PodTemplatesGetter
ReplicationControllersGetter
ResourceQuotasGetter
SecretsGetter
ServicesGetter
ServiceAccountsGetter
}
type CoreV1Client struct {
restClient rest.Interface
}
可以看到
CoreV1Interface定义了corev1下所有资源的获得接口. 而CoreV1Client就是该接口的实现类. 看一个pods方法
// kubernetes/typed/core/v1/core_client.go
func (c *CoreV1Client) Pods(namespace string) PodInterface {
return newPods(c, namespace)
}
可见
Pods方法返回一个可以操作pod的PodInterface. 该PodInterface接口如下所示:
// kubernetes/typed/core/v1/pod.go
type PodInterface interface {
Create(*v1.Pod) (*v1.Pod, error)
Update(*v1.Pod) (*v1.Pod, error)
UpdateStatus(*v1.Pod) (*v1.Pod, error)
Delete(name string, options *metav1.DeleteOptions) error
DeleteCollection(options *metav1.DeleteOptions, listOptions metav1.ListOptions) error
Get(name string, options metav1.GetOptions) (*v1.Pod, error)
List(opts metav1.ListOptions) (*v1.PodList, error)
Watch(opts metav1.ListOptions) (watch.Interface, error)
Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1.Pod, err error)
GetEphemeralContainers(podName string, options metav1.GetOptions) (*v1.EphemeralContainers, error)
UpdateEphemeralContainers(podName string, ephemeralContainers *v1.EphemeralContainers) (*v1.EphemeralContainers, error)
PodExpansion
}
type pods struct {
client rest.Interface
ns string
}
// newPods returns a Pods
func newPods(c *CoreV1Client, namespace string) *pods {
return &pods{
client: c.RESTClient(),
ns: namespace,
}
}
可以看到
pods结构体就是PodInterface的实现类,pods的一个对象表示可以操作该ns这个namespace下面的所有pods. 可以看一下list方法.
// kubernetes/typed/core/v1/pod.go
func (c *pods) List(opts metav1.ListOptions) (result *v1.PodList, err error) {
var timeout time.Duration
if opts.TimeoutSeconds != nil {
timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
}
result = &v1.PodList{}
err = c.client.Get().
Namespace(c.ns).
Resource("pods").
VersionedParams(&opts, scheme.ParameterCodec).
Timeout(timeout).
Do().
Into(result)
return
}
调用
rest client从api-server中获得该namespace下所有的pods.
4. 总结
flow.png
整体的调用关系如上图所示.
