client-go 源码分析（4）

mastertravels77 · 发表于 2023-2-18 07:36

本篇的主题是客户端ClientSet。ClientSet和DynamicClient的优缺点正好互换。ClientSet只能操作内置的资源对象，DynamicClient不仅可以操作内置的资源对象，也可以操作CRD；ClientSet有类型检查，DynamicClient没有。
下面是一个调用ClientSet，查询default namespace下所有pod的例子：
package main

import (
&#34;context&#34;
&#34;fmt&#34;

apiv1 &#34;k8s.io/api/core/v1&#34;
metav1 &#34;k8s.io/apimachinery/pkg/apis/meta/v1&#34;
&#34;k8s.io/client-go/kubernetes&#34;
&#34;k8s.io/client-go/tools/clientcmd&#34;
)

func main() {
// 加载kubeconfig文件，生成config对象
config, err := clientcmd.BuildConfigFromFlags(&#34;&#34;, &#34;C:\\Users\\hanwei\\config&#34;)
if err != nil {
panic(err)
}

// kubernetes.NewForConfig通过config实例化ClientSet对象
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err)
}

//请求core核心资源组v1资源版本下的Pods资源对象
podClient := clientset.CoreV1().Pods(apiv1.NamespaceDefault)
// 设置选项
list, err := podClient.List(context.TODO(), metav1.ListOptions{Limit: 500})
if err != nil {
panic(err)
}

for _, d := range list.Items {
fmt.Printf(&#34;NAMESPACE: %v \t NAME:%v \t STATUS: %+v\n&#34;, d.Namespace, d.Name, d.Status.Phase)
}
}
GOROOT=C:\go\go1.19 #gosetup
GOPATH=C:\Users\hanwei\go #gosetup
C:\go\go1.19\bin\go.exe build -o C:\Users\hanwei\AppData\Local\Temp\GoLand\___2go_build_lab.exe lab #gosetup
C:\Users\hanwei\AppData\Local\Temp\GoLand\___2go_build_lab.exe
NAMESPACE: default    NAME:cdi-upload-windows-2003-001       STATUS: Running
NAMESPACE: default    NAME:tomcat-deployment-5b689c848f-2jprs       STATUS: Running
NAMESPACE: default    NAME:virt-launcher-bc-2003-0907-001-vkd8f    STATUS: Running
NAMESPACE: default    NAME:virt-launcher-test-sg-111-lc9kf STATUS: Running
NAMESPACE: default    NAME:virt-launcher-test-sg-v98xt       STATUS: Running
NAMESPACE: default    NAME:virt-launcher-test-vpc-hxkpx    STATUS: Running
NAMESPACE: default    NAME:virt-launcher-vm-centos-jphml    STATUS: Running

Process finished with the exit code 0上面的main方法核心就是这两句：
//请求core核心资源组v1资源版本下的Pods资源对象
podClient := clientset.CoreV1().Pods(apiv1.NamespaceDefault)
// 设置选项
list, err := podClient.List(context.TODO(), metav1.ListOptions{Limit: 500})main方法中的 clientset.CoreV1() 调用了下面的方法：
func (c *Clientset) CoreV1() corev1.CoreV1Interface {
return c.coreV1
}上面的代码返回的是corev1客户端，值类型是*corev1.CoreV1Client类型，是corev1.CoreV1Interface接口类型的具体类型。
type PodsGetter interface {
Pods(namespace string) PodInterface
}
func (c *CoreV1Client) Pods(namespace string) PodInterface {
return newPods(c, namespace)
}上面的代码可以看出CoreV1Client实现了PodsGetter接口的全部方法，clientset.CoreV1().Pods(apiv1.NamespaceDefault)所以会配到PodsGetter接口的Pods方法，再匹配到具体的func (c *CoreV1Client) Pods方法，并传入namespace参数。
newPods 函数会构造pods结构体，并将函数的形参一个是 corev1 客户端的 RESTClient()方法，就是上篇讲到的restclient客户端，一个是namespace，(c *CoreV1Client, namespace string)分别放进pods结构体的两个属性中。而pods结构体又实现了PodInterface接口的List方法。
main方法中的 clientset.CoreV1().Pods(apiv1.NamespaceDefault) 返回的是PodInterface接口。main方法中的下一句中 podClient.List 实际调用的就是 func (c *pods) List(ctx context.Context, opts metav1.ListOptions) (result *v1.PodList, err error) 方法。
func newPods(c *CoreV1Client, namespace string) *pods {
return &pods{
client: c.RESTClient(),
ns:    namespace,
}
}
type pods struct {
client rest.Interface
ns    string
}
type PodInterface interface {
Create(ctx context.Context, pod *v1.Pod, opts metav1.CreateOptions) (*v1.Pod, error)
Update(ctx context.Context, pod *v1.Pod, opts metav1.UpdateOptions) (*v1.Pod, error)
UpdateStatus(ctx context.Context, pod *v1.Pod, opts metav1.UpdateOptions) (*v1.Pod, error)
Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error
DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error
Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.Pod, error)
List(ctx context.Context, opts metav1.ListOptions) (*v1.PodList, error)
Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)
Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *v1.Pod, err error)
Apply(ctx context.Context, pod *corev1.PodApplyConfiguration, opts metav1.ApplyOptions) (result *v1.Pod, err error)
ApplyStatus(ctx context.Context, pod *corev1.PodApplyConfiguration, opts metav1.ApplyOptions) (result *v1.Pod, err error)
UpdateEphemeralContainers(ctx context.Context, podName string, pod *v1.Pod, opts metav1.UpdateOptions) (*v1.Pod, error)

PodExpansion
}
func (c *pods) List(ctx context.Context, 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(&#34;pods&#34;).
VersionedParams(&opts, scheme.ParameterCodec).
Timeout(timeout).
Do(ctx).
Into(result)
return
}实际上 func (c *pods) List(ctx context.Context, opts metav1.ListOptions) (result *v1.PodList, err error) 方法就是调用上面讲到的 restclient 客户端。参考 client-go 源码分析（3） - rest模块
上面是clientset查询pod list的，下面的删除，创建，都是调用的restclient客户端实现的：
func (c *pods) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
return c.client.Delete().
Namespace(c.ns).
Resource(&#34;pods&#34;).
Name(name).
Body(&opts).
Do(ctx).
Error()
}
func (c *pods) Create(ctx context.Context, pod *v1.Pod, opts metav1.CreateOptions) (result *v1.Pod, err error) {
result = &v1.Pod{}
err = c.client.Post().
Namespace(c.ns).
Resource(&#34;pods&#34;).
VersionedParams(&opts, scheme.ParameterCodec).
Body(pod).
Do(ctx).
Into(result)
return
}
type Pod struct {
metav1.TypeMeta `json:&#34;,inline&#34;`
// Standard object&#39;s metadata.
// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
// +optional
metav1.ObjectMeta `json:&#34;metadata,omitempty&#34; protobuf:&#34;bytes,1,opt,name=metadata&#34;`

// Specification of the desired behavior of the pod.
// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status
// +optional
Spec PodSpec `json:&#34;spec,omitempty&#34; protobuf:&#34;bytes,2,opt,name=spec&#34;`

// Most recently observed status of the pod.
// This data may not be up to date.
// Populated by the system.
// Read-only.
// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status
// +optional
Status PodStatus `json:&#34;status,omitempty&#34; protobuf:&#34;bytes,3,opt,name=status&#34;`
}DynamicClient是一种动态客户端，它可以对任意Kubernetes资源进行RESTful操作，包括CRD自定义资源。DynamicClient与ClientSet操作类似，同样封装了RESTClient，同样提供了Create、Update、Delete、Get、List、Watch、Patch等方法。DynamicClient与ClientSet最大的不同之处是，ClientSet仅能访问Kubernetes自带的资源（即客户端集合内的资源），不能直接访问CRD自定义资源。ClientSet需要预先实现每种Resource和Version的操作，其内部的数据都是结构化数据（即已知数据结构）。而DynamicClient内部实现了Unstructured，用于处理非结构化数据结构（即无法提前预知数据结构），这也是DynamicClient能够处理CRD自定义资源的关键。
注意：

DynamicClient获得的数据都是一个object类型。存的时候是 unstructured
DynamicClient不是类型安全的，因此在访问CRD自定义资源时需要特别注意。例如，在操作指针不当的情况下可能会导致程序崩溃。

package main

import (
&#34;context&#34;
&#34;fmt&#34;

apiv1 &#34;k8s.io/api/core/v1&#34;
corev1 &#34;k8s.io/api/core/v1&#34;
metav1 &#34;k8s.io/apimachinery/pkg/apis/meta/v1&#34;

&#34;k8s.io/apimachinery/pkg/runtime&#34;
&#34;k8s.io/apimachinery/pkg/runtime/schema&#34;
&#34;k8s.io/client-go/dynamic&#34;
&#34;k8s.io/client-go/tools/clientcmd&#34;
)

func main() {
// 加载kubeconfig文件，生成config对象
config, err := clientcmd.BuildConfigFromFlags(&#34;&#34;, &#34;C:\\Users\\hanwei\\config&#34;)
if err != nil {
panic(err)
}

// dynamic.NewForConfig函数通过config实例化dynamicClient对象
dynamicClient, err := dynamic.NewForConfig(config)
if err != nil {
panic(err)
}

// 通过schema.GroupVersionResource设置请求的资源版本和资源组，设置命名空间和请求参数,得到unstructured.UnstructuredList指针类型的PodList
gvr := schema.GroupVersionResource{Version: &#34;v1&#34;, Resource: &#34;pods&#34;}
unstructObj, err := dynamicClient.Resource(gvr).Namespace(apiv1.NamespaceDefault).List(context.TODO(), metav1.ListOptions{Limit: 500})
if err != nil {
panic(err)
}

// 通过runtime.DefaultUnstructuredConverter函数将unstructured.UnstructuredList转为PodList类型
podList := &corev1.PodList{}
err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructObj.UnstructuredContent(), podList)
if err != nil {
panic(err)
}

for _, d := range podList.Items {
fmt.Printf(&#34;NAMESPACE: %v NAME:%v \t STATUS: %+v\n&#34;, d.Namespace, d.Name, d.Status.Phase)
}
}
GOROOT=C:\go\go1.19 #gosetup
GOPATH=C:\Users\hanwei\go #gosetup
C:\go\go1.19\bin\go.exe build -o C:\Users\hanwei\AppData\Local\Temp\GoLand\___4go_build_lab.exe lab #gosetup
C:\Users\hanwei\AppData\Local\Temp\GoLand\___4go_build_lab.exe
NAMESPACE: default NAME:cdi-upload-windows-2003-001    STATUS: Running
NAMESPACE: default NAME:hp-volume-7lvp4       STATUS: Running
NAMESPACE: default NAME:tomcat-deployment-5b689c848f-2jprs    STATUS: Running
NAMESPACE: default NAME:virt-launcher-bc-2003-0907-001-vkd8f    STATUS: Running
NAMESPACE: default NAME:virt-launcher-test-sg-111-lc9kf       STATUS: Running
NAMESPACE: default NAME:virt-launcher-test-sg-v98xt    STATUS: Running
NAMESPACE: default NAME:virt-launcher-test-snapshot-v-z52fv    STATUS: Running
NAMESPACE: default NAME:virt-launcher-test-vpc-q8rwz    STATUS: Running
NAMESPACE: default NAME:virt-launcher-vm-centos-q5zq5 STATUS: Running

Process finished with the exit code 0和ClientSet客户端一样，调用的ResetClient客户端。不同的是将response body的数据转成非结构化的数据体返回。
func (c *dynamicResourceClient) List(ctx context.Context, opts metav1.ListOptions) (*unstructured.UnstructuredList, error) {
result := c.client.client.Get().AbsPath(c.makeURLSegments(&#34;&#34;)...).SpecificallyVersionedParams(&opts, dynamicParameterCodec, versionV1).Do(ctx)
if err := result.Error(); err != nil {
return nil, err
}
retBytes, err := result.Raw()
if err != nil {
return nil, err
}
uncastObj, err := runtime.Decode(unstructured.UnstructuredJSONScheme, retBytes)
if err != nil {
return nil, err
}
if list, ok := uncastObj.(*unstructured.UnstructuredList); ok {
return list, nil
}

list, err := uncastObj.(*unstructured.Unstructured).ToList()
if err != nil {
return nil, err
}
return list, nil
}在调用 runtime.DefaultUnstructuredConverter.FromUnstructured 将上面的非结构化数据转换成Kubernetes资源对象数据类型（使用encoding/json/Unmarshaler进行转换，若无法通过上述方式转换，用反射机制进行转换）。

欢迎订阅微信公众号“后端云”！

		自动登录	找回密码
密码			立即注册