client-go的使用及源码分析

client-go^[1]是Kubernetes 官方团队支持的Go语言客户端库。

示例代码

package main

import (
    "context"
    "flag"
    "fmt"
    "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"
    "k8s.io/client-go/util/homedir"
)

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)
    }
}

client-go源码分析

client-go源码：https://github.com/kubernetes/client-go

源码目录结构：

• The kubernetes package contains the clientset to access Kubernetes API.
• The discovery package is used to discover APIs supported by a Kubernetes API server.
• The dynamic package contains a dynamic client that can perform generic operations on arbitrary Kubernetes API objects.
• The plugin/pkg/client/auth packages contain optional authentication plugins for obtaining credentials from external sources.
• The transportpackage is used to set up auth and start a connection.
• The tools/cache package is useful for writing controllers.

kubeconfig

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")
}

restclient.Config

BuildConfigFromFlags

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()
}

clientset

clientset, err := kubernetes.NewForConfig(config)

NewForConfig

func NewForConfig(c *rest.Config) (*Clientset, error) {
    configShallowCopy := *c

    if configShallowCopy.UserAgent == "" {
        configShallowCopy.UserAgent = rest.DefaultKubernetesUserAgent()
    }

    // share the transport between all clients
    httpClient, err := rest.HTTPClientFor(&configShallowCopy)
    if err != nil {
        return nil, err
    }

    return NewForConfigAndClient(&configShallowCopy, httpClient)
}

clientset的结构体

// Clientset contains the clients for groups. Each group has exactly one
// version included in a Clientset.
type Clientset struct {
    *discovery.DiscoveryClient
    admissionregistrationV1      *admissionregistrationv1.AdmissionregistrationV1Client
    admissionregistrationV1beta1 *admissionregistrationv1beta1.AdmissionregistrationV1beta1Client
    internalV1alpha1             *internalv1alpha1.InternalV1alpha1Client
    appsV1                       *appsv1.AppsV1Client
    appsV1beta1                  *appsv1beta1.AppsV1beta1Client
    appsV1beta2                  *appsv1beta2.AppsV1beta2Client
    authenticationV1             *authenticationv1.AuthenticationV1Client
    authenticationV1beta1        *authenticationv1beta1.AuthenticationV1beta1Client
    authorizationV1              *authorizationv1.AuthorizationV1Client
    authorizationV1beta1         *authorizationv1beta1.AuthorizationV1beta1Client
    autoscalingV1                *autoscalingv1.AutoscalingV1Client
    autoscalingV2                *autoscalingv2.AutoscalingV2Client
    autoscalingV2beta1           *autoscalingv2beta1.AutoscalingV2beta1Client
    autoscalingV2beta2           *autoscalingv2beta2.AutoscalingV2beta2Client
    batchV1                      *batchv1.BatchV1Client
    batchV1beta1                 *batchv1beta1.BatchV1beta1Client
    certificatesV1               *certificatesv1.CertificatesV1Client
    certificatesV1beta1          *certificatesv1beta1.CertificatesV1beta1Client
    coordinationV1beta1          *coordinationv1beta1.CoordinationV1beta1Client
    coordinationV1               *coordinationv1.CoordinationV1Client
    coreV1                       *corev1.CoreV1Client
    discoveryV1                  *discoveryv1.DiscoveryV1Client
    discoveryV1beta1             *discoveryv1beta1.DiscoveryV1beta1Client
    eventsV1                     *eventsv1.EventsV1Client
    eventsV1beta1                *eventsv1beta1.EventsV1beta1Client
    extensionsV1beta1            *extensionsv1beta1.ExtensionsV1beta1Client
    flowcontrolV1alpha1          *flowcontrolv1alpha1.FlowcontrolV1alpha1Client
    flowcontrolV1beta1           *flowcontrolv1beta1.FlowcontrolV1beta1Client
    flowcontrolV1beta2           *flowcontrolv1beta2.FlowcontrolV1beta2Client
    networkingV1                 *networkingv1.NetworkingV1Client
    networkingV1beta1            *networkingv1beta1.NetworkingV1beta1Client
    nodeV1                       *nodev1.NodeV1Client
    nodeV1alpha1                 *nodev1alpha1.NodeV1alpha1Client
    nodeV1beta1                  *nodev1beta1.NodeV1beta1Client
    policyV1                     *policyv1.PolicyV1Client
    policyV1beta1                *policyv1beta1.PolicyV1beta1Client
    rbacV1                       *rbacv1.RbacV1Client
    rbacV1beta1                  *rbacv1beta1.RbacV1beta1Client
    rbacV1alpha1                 *rbacv1alpha1.RbacV1alpha1Client
    schedulingV1alpha1           *schedulingv1alpha1.SchedulingV1alpha1Client
    schedulingV1beta1            *schedulingv1beta1.SchedulingV1beta1Client
    schedulingV1                 *schedulingv1.SchedulingV1Client
    storageV1beta1               *storagev1beta1.StorageV1beta1Client
    storageV1                    *storagev1.StorageV1Client
    storageV1alpha1              *storagev1alpha1.StorageV1alpha1Client
}

clientset.Interface

type Interface interface {
    Discovery() discovery.DiscoveryInterface
    AdmissionregistrationV1() admissionregistrationv1.AdmissionregistrationV1Interface
    AdmissionregistrationV1beta1() admissionregistrationv1beta1.AdmissionregistrationV1beta1Interface
    InternalV1alpha1() internalv1alpha1.InternalV1alpha1Interface
    AppsV1() appsv1.AppsV1Interface
    AppsV1beta1() appsv1beta1.AppsV1beta1Interface
    AppsV1beta2() appsv1beta2.AppsV1beta2Interface
    AuthenticationV1() authenticationv1.AuthenticationV1Interface
    AuthenticationV1beta1() authenticationv1beta1.AuthenticationV1beta1Interface
    AuthorizationV1() authorizationv1.AuthorizationV1Interface
    AuthorizationV1beta1() authorizationv1beta1.AuthorizationV1beta1Interface
    AutoscalingV1() autoscalingv1.AutoscalingV1Interface
    AutoscalingV2() autoscalingv2.AutoscalingV2Interface
    AutoscalingV2beta1() autoscalingv2beta1.AutoscalingV2beta1Interface
    AutoscalingV2beta2() autoscalingv2beta2.AutoscalingV2beta2Interface
    BatchV1() batchv1.BatchV1Interface
    BatchV1beta1() batchv1beta1.BatchV1beta1Interface
    CertificatesV1() certificatesv1.CertificatesV1Interface
    CertificatesV1beta1() certificatesv1beta1.CertificatesV1beta1Interface
    CoordinationV1beta1() coordinationv1beta1.CoordinationV1beta1Interface
    CoordinationV1() coordinationv1.CoordinationV1Interface
    CoreV1() corev1.CoreV1Interface
    DiscoveryV1() discoveryv1.DiscoveryV1Interface
    DiscoveryV1beta1() discoveryv1beta1.DiscoveryV1beta1Interface
    EventsV1() eventsv1.EventsV1Interface
    EventsV1beta1() eventsv1beta1.EventsV1beta1Interface
    ExtensionsV1beta1() extensionsv1beta1.ExtensionsV1beta1Interface
    FlowcontrolV1alpha1() flowcontrolv1alpha1.FlowcontrolV1alpha1Interface
    FlowcontrolV1beta1() flowcontrolv1beta1.FlowcontrolV1beta1Interface
    FlowcontrolV1beta2() flowcontrolv1beta2.FlowcontrolV1beta2Interface
    NetworkingV1() networkingv1.NetworkingV1Interface
    NetworkingV1beta1() networkingv1beta1.NetworkingV1beta1Interface
    NodeV1() nodev1.NodeV1Interface
    NodeV1alpha1() nodev1alpha1.NodeV1alpha1Interface
    NodeV1beta1() nodev1beta1.NodeV1beta1Interface
    PolicyV1() policyv1.PolicyV1Interface
    PolicyV1beta1() policyv1beta1.PolicyV1beta1Interface
    RbacV1() rbacv1.RbacV1Interface
    RbacV1beta1() rbacv1beta1.RbacV1beta1Interface
    RbacV1alpha1() rbacv1alpha1.RbacV1alpha1Interface
    SchedulingV1alpha1() schedulingv1alpha1.SchedulingV1alpha1Interface
    SchedulingV1beta1() schedulingv1beta1.SchedulingV1beta1Interface
    SchedulingV1() schedulingv1.SchedulingV1Interface
    StorageV1beta1() storagev1beta1.StorageV1beta1Interface
    StorageV1() storagev1.StorageV1Interface
    StorageV1alpha1() storagev1alpha1.StorageV1alpha1Interface
}

CoreV1Client

我们以clientset中的CoreV1Client为例做分析

cs.coreV1, err = corev1.NewForConfigAndClient(&configShallowCopy, httpClient)

corev1.NewForConfigAndClient

// NewForConfigAndClient creates a new CoreV1Client for the given config and http client.
// Note the http client provided takes precedence over the configured transport values.
func NewForConfigAndClient(c *rest.Config, h *http.Client) (*CoreV1Client, error) {
    config := *c
    if err := setConfigDefaults(&config); err != nil {
        return nil, err
    }
    client, err := rest.RESTClientForConfigAndClient(&config, h)
    if err != nil {
        return nil, err
    }
    return &CoreV1Client{client}, nil
}

CoreV1Client结构体

type CoreV1Client struct {
    restClient rest.Interface
}

CoreV1Client实现了CoreV1Interface的接口，即以下方法，从而对kubernetes的资源对象进行增删改查的操作。

func (c *CoreV1Client) ComponentStatuses() ComponentStatusInterface {
    return newComponentStatuses(c)
}

func (c *CoreV1Client) ConfigMaps(namespace string) ConfigMapInterface {
    return newConfigMaps(c, namespace)
}

func (c *CoreV1Client) Endpoints(namespace string) EndpointsInterface {
    return newEndpoints(c, namespace)
}

func (c *CoreV1Client) Events(namespace string) EventInterface {
    return newEvents(c, namespace)
}

func (c *CoreV1Client) LimitRanges(namespace string) LimitRangeInterface {
    return newLimitRanges(c, namespace)
}

func (c *CoreV1Client) Namespaces() NamespaceInterface {
    return newNamespaces(c)
}

func (c *CoreV1Client) Nodes() NodeInterface {
    return newNodes(c)
}

func (c *CoreV1Client) PersistentVolumes() PersistentVolumeInterface {
    return newPersistentVolumes(c)
}

func (c *CoreV1Client) PersistentVolumeClaims(namespace string) PersistentVolumeClaimInterface {
    return newPersistentVolumeClaims(c, namespace)
}

func (c *CoreV1Client) Pods(namespace string) PodInterface {
    return newPods(c, namespace)
}

func (c *CoreV1Client) PodTemplates(namespace string) PodTemplateInterface {
    return newPodTemplates(c, namespace)
}

func (c *CoreV1Client) ReplicationControllers(namespace string) ReplicationControllerInterface {
    return newReplicationControllers(c, namespace)
}

func (c *CoreV1Client) ResourceQuotas(namespace string) ResourceQuotaInterface {
    return newResourceQuotas(c, namespace)
}

func (c *CoreV1Client) Secrets(namespace string) SecretInterface {
    return newSecrets(c, namespace)
}

func (c *CoreV1Client) Services(namespace string) ServiceInterface {
    return newServices(c, namespace)
}

func (c *CoreV1Client) ServiceAccounts(namespace string) ServiceAccountInterface {
    return newServiceAccounts(c, namespace)
}

CoreV1Interface

type CoreV1Interface interface {
    RESTClient() rest.Interface
    ComponentStatusesGetter
    ConfigMapsGetter
    EndpointsGetter
    EventsGetter
    LimitRangesGetter
    NamespacesGetter
    NodesGetter
    PersistentVolumesGetter
    PersistentVolumeClaimsGetter
    PodsGetter
    PodTemplatesGetter
    ReplicationControllersGetter
    ResourceQuotasGetter
    SecretsGetter
    ServicesGetter
    ServiceAccountsGetter
}

CoreV1Interface中包含了各种kubernetes对象的调用接口，例如PodsGetter是对kubernetes中pod对象增删改查操作的接口。ServicesGetter是对service对象的操作的接口。

PodsGetter

以PodsGetter接口为例分析CoreV1Client对pod对象的增删改查接口调用。

pods, err := clientset.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{})

CoreV1().Pods()

func (c *CoreV1Client) Pods(namespace string) PodInterface {
    return newPods(c, namespace)
}

newPods()

// newPods returns a Pods
func newPods(c *CoreV1Client, namespace string) *pods {
    return &pods{
        client: c.RESTClient(),
        ns:     namespace,
    }
}

// pods implements PodInterface
type pods struct {
    client rest.Interface
    ns     string
}

// PodInterface has methods to work with Pod resources.
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
}

PodsGetter

// PodsGetter has a method to return a PodInterface.
// A group's client should implement this interface.
type PodsGetter interface {
    Pods(namespace string) PodInterface
}

Pods().List()

// List takes label and field selectors, and returns the list of Pods that match those selectors.
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("pods").
        VersionedParams(&opts, scheme.ParameterCodec).
        Timeout(timeout).
        Do(ctx).
        Into(result)
    return
}

RESTClient

client, err := rest.RESTClientForConfigAndClient(&config, h)

rest.RESTClientForConfigAndClient

func RESTClientForConfigAndClient(config *Config, httpClient *http.Client) (*RESTClient, error) {
    if config.GroupVersion == nil {
        return nil, fmt.Errorf("GroupVersion is required when initializing a RESTClient")
    }
    if config.NegotiatedSerializer == nil {
        return nil, fmt.Errorf("NegotiatedSerializer is required when initializing a RESTClient")
    }

    baseURL, versionedAPIPath, err := defaultServerUrlFor(config)
    if err != nil {
        return nil, err
    }

    rateLimiter := config.RateLimiter
    if rateLimiter == nil {
        qps := config.QPS
        if config.QPS == 0.0 {
            qps = DefaultQPS
        }
        burst := config.Burst
        if config.Burst == 0 {
            burst = DefaultBurst
        }
        if qps > 0 {
            rateLimiter = flowcontrol.NewTokenBucketRateLimiter(qps, burst)
        }
    }

    var gv schema.GroupVersion
    if config.GroupVersion != nil {
        gv = *config.GroupVersion
    }
    clientContent := ClientContentConfig{
        AcceptContentTypes: config.AcceptContentTypes,
        ContentType:        config.ContentType,
        GroupVersion:       gv,
        Negotiator:         runtime.NewClientNegotiator(config.NegotiatedSerializer, gv),
    }

    restClient, err := NewRESTClient(baseURL, versionedAPIPath, clientContent, rateLimiter, httpClient)
    if err == nil && config.WarningHandler != nil {
        restClient.warningHandler = config.WarningHandler
    }
    return restClient, err
}

NewRESTClient

func NewRESTClient(baseURL *url.URL, versionedAPIPath string, config ClientContentConfig, rateLimiter flowcontrol.RateLimiter, client *http.Client) (*RESTClient, error) {
    if len(config.ContentType) == 0 {
        config.ContentType = "application/json"
    }

    base := *baseURL
    if !strings.HasSuffix(base.Path, "/") {
        base.Path += "/"
    }
    base.RawQuery = ""
    base.Fragment = ""

    return &RESTClient{
        base:             &base,
        versionedAPIPath: versionedAPIPath,
        content:          config,
        createBackoffMgr: readExpBackoffConfig,
        rateLimiter:      rateLimiter,

        Client: client,
    }, nil
}

RESTClient结构体

type RESTClient struct {
    // base is the root URL for all invocations of the client
    base *url.URL
    // versionedAPIPath is a path segment connecting the base URL to the resource root
    versionedAPIPath string

    // content describes how a RESTClient encodes and decodes responses.
    content ClientContentConfig

    // creates BackoffManager that is passed to requests.
    createBackoffMgr func() BackoffManager

    // rateLimiter is shared among all requests created by this client unless specifically
    // overridden.
    rateLimiter flowcontrol.RateLimiter

    // warningHandler is shared among all requests created by this client.
    // If not set, defaultWarningHandler is used.
    warningHandler WarningHandler

    // Set specific behavior of the client.  If not set http.DefaultClient will be used.
    Client *http.Client
}

RESTClient.Interface

// Interface captures the set of operations for generically interacting with Kubernetes REST apis.
type Interface interface {
    GetRateLimiter() flowcontrol.RateLimiter
    Verb(verb string) *Request
    Post() *Request
    Put() *Request
    Patch(pt types.PatchType) *Request
    Get() *Request
    Delete() *Request
    APIVersion() schema.GroupVersion
}

在调用HTTP方法（Post()，Put()，Get()，Delete() ）时，实际上调用了Verb(verb string)函数。

// Verb begins a request with a verb (GET, POST, PUT, DELETE).
//
// Example usage of RESTClient's request building interface:
// c, err := NewRESTClient(...)
// if err != nil { ... }
// resp, err := c.Verb("GET").
//  Path("pods").
//  SelectorParam("labels", "area=staging").
//  Timeout(10*time.Second).
//  Do()
// if err != nil { ... }
// list, ok := resp.(*api.PodList)
//
func (c *RESTClient) Verb(verb string) *Request {
    return NewRequest(c).Verb(verb)
}

Verb函数调用了NewRequest方法，最后调用Do()方法实现一个HTTP请求获取Result。

总结

整个调用的过程如下：
kubeconfig→rest.config→clientset→具体的client(CoreV1Client)→具体的资源对象(pod)→RESTClient→http.Client→HTTP请求的发送及响应

参考资料：
[1]：https://kubernetes.io/zh-cn/docs/reference/using-api/client-libraries/

原文链接：https://www.huweihuang.com/article/source-analysis/client-go-source-analysis/