本篇是client-go源码分析的第一篇,client-go是从事Kubernetes开发必研究的项目,client-go之所以重要,主要在以下几个方面:
- 是调用Kubernetes API的唯一的golang官方库。
- 在Kubernetes源码中,如果Kubernetes的某个组件,需要List/Get Kubernetes中的Object,在绝大多 数情况下,会直接使用client-go中的Informer实例中的Lister()方法(该方法包含 了 Get 和 List 方法),而很少直接请求Kubernetes API。
- 用operator sdk开发自己的crd和custom controller开发,必须要用到的golang库,因为kubernetes易扩展的架构设计,只要是基于kubernetes二次开发的项目,基本都要开发自己定制的crd和controller。
本篇从相对简单的discovery包开始:
discovery包的主要作用就是提供当前k8s集群支持哪些资源以及版本信息。
Kubernetes API Server暴露出/api和/apis接口。DiscoveryClient通过RESTClient分别请求/api和/apis接口,从而获取Kubernetes API Server所支持的资源组、资源版信息。这个是通过ServerGroups函数实现的,有了group, version信息后,但是还是不够,因为还没有具体到资源。ServerGroupsAndResources 就获得了所有的资源信息(所有的GVR资源信息),而在Resource资源的定义中,会定义好该资源支持哪些操作:list, delelte ,get等等。
所以kubectl中就使用discovery做了资源的校验。获取所有资源的版本信息,以及支持的操作。就可以判断客户端当前操作是否合理。
先上调用discovery包获取所有gv和gvr信息的代码:
代码语言:javascript复制package main
import (
"fmt"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/discovery"
"k8s.io/client-go/tools/clientcmd"
)
func main() {
// 加载kubeconfig文件,生成config对象
config, err := clientcmd.BuildConfigFromFlags("", "C:\Users\hanwei\config")
if err != nil {
panic(err)
}
// discovery.NewDiscoveryClientForConfigg函数通过config实例化discoveryClient对象
discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
if err != nil {
panic(err)
}
// discoveryClient.ServerGroupsAndResources 返回API Server所支持的资源组、资源版本、资源信息
_, APIResourceList, err := discoveryClient.ServerGroupsAndResources()
if err != nil {
panic(err)
}
// 输出所有资源信息
for _, list := range APIResourceList {
gv, err := schema.ParseGroupVersion(list.GroupVersion)
if err != nil {
panic(err)
}
for _, resource := range list.APIResources {
fmt.Printf("NAME: %v, GROUP: %v, VERSION: %v n", resource.Name, gv.Group, gv.Version)
}
}
}language-go复制代码
代码语言:javascript复制OROOT=C:gogo1.19 #gosetup
GOPATH=C:Usershanweigo #gosetup
C:gogo1.19bingo.exe build -o C:UsershanweiAppDataLocalTempGoLand___go_build_main_go.exe C:UsershanweiDocumentsGoProjectlablab01main.go #gosetup
C:UsershanweiAppDataLocalTempGoLand___go_build_main_go.exe
... // 省略
NAME: pods, GROUP: , VERSION: v1
NAME: pods/attach, GROUP: , VERSION: v1
NAME: pods/binding, GROUP: , VERSION: v1
NAME: pods/eviction, GROUP: , VERSION: v1
NAME: pods/exec, GROUP: , VERSION: v1
NAME: pods/log, GROUP: , VERSION: v1
NAME: pods/portforward, GROUP: , VERSION: v1
NAME: pods/proxy, GROUP: , VERSION: v1
NAME: pods/status, GROUP: , VERSION: v1
NAME: challenges/status, GROUP: acme.cert-manager.io, VERSION: v1
NAME: cephbucketnotifications, GROUP: ceph.rook.io, VERSION: v1
NAME: cephbucketnotifications/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephnfses, GROUP: ceph.rook.io, VERSION: v1
NAME: cephnfses/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystemsubvolumegroups, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystemsubvolumegroups/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystemmirrors, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystemmirrors/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephblockpools, GROUP: ceph.rook.io, VERSION: v1
NAME: cephblockpools/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectstores, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectstores/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephclusters, GROUP: ceph.rook.io, VERSION: v1
NAME: cephclusters/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectzones, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectzones/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephrbdmirrors, GROUP: ceph.rook.io, VERSION: v1
NAME: cephrbdmirrors/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephblockpoolradosnamespaces, GROUP: ceph.rook.io, VERSION: v1
NAME: cephblockpoolradosnamespaces/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectrealms, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectrealms/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephclients, GROUP: ceph.rook.io, VERSION: v1
NAME: cephclients/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectstoreusers, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectstoreusers/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystems, GROUP: ceph.rook.io, VERSION: v1
NAME: cephfilesystems/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectzonegroups, GROUP: ceph.rook.io, VERSION: v1
NAME: cephobjectzonegroups/status, GROUP: ceph.rook.io, VERSION: v1
NAME: cephbuckettopics, GROUP: ceph.rook.io, VERSION: v1
NAME: cephbuckettopics/status, GROUP: ceph.rook.io, VERSION: v1
NAME: virtualmachineinstancereplicasets, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancereplicasets/status, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancereplicasets/scale, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancepresets, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancemigrations, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancemigrations/status, GROUP: kubevirt.io, VERSION: v1
NAME: kubevirts, GROUP: kubevirt.io, VERSION: v1
NAME: kubevirts/status, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstances, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachines, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachines/status, GROUP: kubevirt.io, VERSION: v1
NAME: virtualmachineinstancereplicasets, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstancereplicasets/status, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstancereplicasets/scale, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstancepresets, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstancemigrations, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstancemigrations/status, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: kubevirts, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: kubevirts/status, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachineinstances, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachines, GROUP: kubevirt.io, VERSION: v1alpha3
NAME: virtualmachines/status, GROUP: kubevirt.io, VERSION: v1alpha3
... // 省略
Process finished with the exit code 0language-bash复制代码
kubectl api-resources 和 kubectl api-versions 这两条命令也是调用的该discover包。
代码语言:javascript复制[root@k8s-11 .kube]# kubectl |grep api
api-resources Print the supported API resources on the server
api-versions Print the supported API versions on the server, in the form of "group/version"language-bash复制代码
上面的main方法,分为4步:
step1:clientcmd.BuildConfigFromFlags 函数返回 config对象,这里用了kubeconfig的本地路径。
step2:discovery.NewDiscoveryClientForConfig 函数返回 DiscoveryClient对象。
代码语言:javascript复制func BuildConfigFromFlags(masterUrl, kubeconfigPath string) (*restclient.Config, error) {
if kubeconfigPath == "" && masterUrl == "" {
klog.Warning("Neither --kubeconfig nor --master was specified. Using the inClusterConfig. This might not work.")
kubeconfig, err := restclient.InClusterConfig()
if err == nil {
return kubeconfig, nil
}
klog.Warning("error creating inClusterConfig, falling back to default config: ", err)
}
return NewNonInteractiveDeferredLoadingClientConfig(
&ClientConfigLoadingRules{ExplicitPath: kubeconfigPath},
&ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: masterUrl}}).ClientConfig()
}language-go复制代码
代码语言:javascript复制func NewDiscoveryClientForConfig(c *restclient.Config) (*DiscoveryClient, error) {
config := *c
if err := setDiscoveryDefaults(&config); err != nil {
return nil, err
}
httpClient, err := restclient.HTTPClientFor(&config)
if err != nil {
return nil, err
}
return NewDiscoveryClientForConfigAndClient(&config, httpClient)
}language-go复制代码
step3:discoveryClient.ServerGroupsAndResources 函数相对绕一点。
代码语言:javascript复制func (d *DiscoveryClient) ServerGroupsAndResources() ([]*metav1.APIGroup, []*metav1.APIResourceList, error) {
return withRetries(defaultRetries, func() ([]*metav1.APIGroup, []*metav1.APIResourceList, error) {
return ServerGroupsAndResources(d)
})
}language-go复制代码
withRetries 函数 多次尝试执行 函数 ServerGroupsAndResources(d DiscoveryInterface) ([]*metav1.APIGroup, []*metav1.APIResourceList, error) ,该函数的形式参数是interface,用到了golang中的多态。因为实参是结构体DiscoveryClient,所以多态会调用DiscoveryClient下对应的具体方法。
代码语言:javascript复制func ServerGroupsAndResources(d DiscoveryInterface) ([]*metav1.APIGroup, []*metav1.APIResourceList, error) {
sgs, err := d.ServerGroups()
if sgs == nil {
return nil, nil, err
}
resultGroups := []*metav1.APIGroup{}
for i := range sgs.Groups {
resultGroups = append(resultGroups, &sgs.Groups[i])
}
groupVersionResources, failedGroups := fetchGroupVersionResources(d, sgs)
// order results by group/version discovery order
result := []*metav1.APIResourceList{}
for _, apiGroup := range sgs.Groups {
for _, version := range apiGroup.Versions {
gv := schema.GroupVersion{Group: apiGroup.Name, Version: version.Version}
if resources, ok := groupVersionResources[gv]; ok {
result = append(result, resources)
}
}
}
if len(failedGroups) == 0 {
return resultGroups, result, nil
}
return resultGroups, result, &ErrGroupDiscoveryFailed{Groups: failedGroups}
}language-go复制代码
ServerGroupsAndResources(d DiscoveryInterface) ([]*metav1.APIGroup, []*metav1.APIResourceList, error) 主要就是 调用了 d.ServerGroups() 以及 fetchGroupVersionResources(d, sgs) 两个方法,分别获取group信息和resource信息。
代码语言:javascript复制type ServerGroupsInterface interface {
// ServerGroups returns the supported groups, with information like supported versions and the
// preferred version.
ServerGroups() (*metav1.APIGroupList, error)
}language-go复制代码
代码语言:javascript复制func (d *DiscoveryClient) ServerGroups() (apiGroupList *metav1.APIGroupList, err error) {
// Get the groupVersions exposed at /api
v := &metav1.APIVersions{}
err = d.restClient.Get().AbsPath(d.LegacyPrefix).Do(context.TODO()).Into(v)
apiGroup := metav1.APIGroup{}
if err == nil && len(v.Versions) != 0 {
apiGroup = apiVersionsToAPIGroup(v)
}
if err != nil && !errors.IsNotFound(err) && !errors.IsForbidden(err) {
return nil, err
}
// Get the groupVersions exposed at /apis
apiGroupList = &metav1.APIGroupList{}
err = d.restClient.Get().AbsPath("/apis").Do(context.TODO()).Into(apiGroupList)
if err != nil && !errors.IsNotFound(err) && !errors.IsForbidden(err) {
return nil, err
}
// to be compatible with a v1.0 server, if it's a 403 or 404, ignore and return whatever we got from /api
if err != nil && (errors.IsNotFound(err) || errors.IsForbidden(err)) {
apiGroupList = &metav1.APIGroupList{}
}
// prepend the group retrieved from /api to the list if not empty
if len(v.Versions) != 0 {
apiGroupList.Groups = append([]metav1.APIGroup{apiGroup}, apiGroupList.Groups...)
}
return apiGroupList, nil
}language-go复制代码
func (d *DiscoveryClient) ServerGroups() (apiGroupList *metav1.APIGroupList, err error) 调用restClient Get方法访问Kubernetes API,获取apiGroup信息,调用了两次,分别是历史遗留的/api接口,对应corev1,还有就是有gv具体名称的/apis接口。
代码语言:javascript复制func fetchGroupVersionResources(d DiscoveryInterface, apiGroups *metav1.APIGroupList) (map[schema.GroupVersion]*metav1.APIResourceList, map[schema.GroupVersion]error) {
groupVersionResources := make(map[schema.GroupVersion]*metav1.APIResourceList)
failedGroups := make(map[schema.GroupVersion]error)
wg := &sync.WaitGroup{}
resultLock := &sync.Mutex{}
for _, apiGroup := range apiGroups.Groups {
for _, version := range apiGroup.Versions {
groupVersion := schema.GroupVersion{Group: apiGroup.Name, Version: version.Version}
wg.Add(1)
go func() {
defer wg.Done()
defer utilruntime.HandleCrash()
apiResourceList, err := d.ServerResourcesForGroupVersion(groupVersion.String())
// lock to record results
resultLock.Lock()
defer resultLock.Unlock()
if err != nil {
// TODO: maybe restrict this to NotFound errors
failedGroups[groupVersion] = err
}
if apiResourceList != nil {
// even in case of error, some fallback might have been returned
groupVersionResources[groupVersion] = apiResourceList
}
}()
}
}
wg.Wait()
return groupVersionResources, failedGroups
}language-go复制代码
代码语言:javascript复制func (d *DiscoveryClient) ServerResourcesForGroupVersion(groupVersion string) (resources *metav1.APIResourceList, err error) {
url := url.URL{}
if len(groupVersion) == 0 {
return nil, fmt.Errorf("groupVersion shouldn't be empty")
}
if len(d.LegacyPrefix) > 0 && groupVersion == "v1" {
url.Path = d.LegacyPrefix "/" groupVersion
} else {
url.Path = "/apis/" groupVersion
}
resources = &metav1.APIResourceList{
GroupVersion: groupVersion,
}
err = d.restClient.Get().AbsPath(url.String()).Do(context.TODO()).Into(resources)
if err != nil {
// ignore 403 or 404 error to be compatible with an v1.0 server.
if groupVersion == "v1" && (errors.IsNotFound(err) || errors.IsForbidden(err)) {
return resources, nil
}
return nil, err
}
return resources, nil
}language-go复制代码
因为获取gv对应的resouce信息需要每一个gv都需要发送一个rest请求,大量rest请求比较耗时,为了加快速度,fetchGroupVersionResources 方法采取go协程同时并发的方式发送rest api请求。
step4:最后将查到的变量APIResourceList通过两层for循环输出所有的resouce信息。