想学习K8S源码,可以加 :mkjnnm
1、介绍
client-go 是用来和 k8s 集群交互的go语言客户端库,地址为:https://github.com/kubernetes/client-go client-go 的版本有两种标识方式:
- v0.x.y (For each
v1.x.y
Kubernetes release, the major version (first digit) would remain0
) - kubernetes-1.x.y
ps: k8s 发布v1.30.0 版本时,client-go 会同步发布两个版本 v0.30.0 (推荐使用)和 kubernetes-1.30.0
如果你想使用最新版本的客户端库,你可以运行如下命令:
go get k8s.io/client-go@latest
如果你想使用特定版本,推荐使用如下命令:
go get k8s.io/client-go@v0.20.4
2、目录结构
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
transport
package is used to set up auth and start a connection. - The
tools/cache
package is useful for writing controllers.
3、使用
1.初始化
- 集群内初始化
- 集群外初始化
如果你的应用跑在集群中的pod上,推荐使用集群内初始化,否则使用集群外初始化。
// 集群外初始化 client
// 需要集群的 kubeconfig 配置文件
func main() {var kubeconfig *stringif 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 kubeconfigconfig, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)if err != nil {panic(err.Error())}// create the clientsetclientset, err := kubernetes.NewForConfig(config)if err != nil {panic(err.Error())}
}
2. 列出集群内pod资源
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))
上述代码在client-
go仓库下examples目录
你可以运行该代码:
cd out-of-cluster-client-configuration
go build -o app .
./app -kubeconfig=/path/to/xxx
3. deployment资源CRUD - typed (clientset)方式
以下代码创建一个有2个副本数的deployment,然后更新副本数为1,升级nginx镜像版本,最后删除
deploymentsClient := clientset.AppsV1().Deployments(apiv1.NamespaceDefault)deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "demo-deployment",},Spec: appsv1.DeploymentSpec{Replicas: int32Ptr(2),Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "demo",},},Template: apiv1.PodTemplateSpec{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "demo",},},Spec: apiv1.PodSpec{Containers: []apiv1.Container{{Name: "web",Image: "nginx:1.12",Ports: []apiv1.ContainerPort{{Name: "http",Protocol: apiv1.ProtocolTCP,ContainerPort: 80,},},},},},},},}// Create Deploymentfmt.Println("Creating deployment...")result, err := deploymentsClient.Create(context.TODO(), deployment, metav1.CreateOptions{})if err != nil {panic(err)}fmt.Printf("Created deployment %q.\n", result.GetObjectMeta().GetName())// Update Deploymentprompt()fmt.Println("Updating deployment...")// You have two options to Update() this Deployment://// 1. Modify the "deployment" variable and call: Update(deployment).// This works like the "kubectl replace" command and it overwrites/loses changes// made by other clients between you Create() and Update() the object.// 2. Modify the "result" returned by Get() and retry Update(result) until// you no longer get a conflict error. This way, you can preserve changes made// by other clients between Create() and Update(). This is implemented below// using the retry utility package included with client-go. (RECOMMENDED)//// More Info:// https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#concurrency-control-and-consistencyretryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {// Retrieve the latest version of Deployment before attempting update// RetryOnConflict uses exponential backoff to avoid exhausting the apiserverresult, getErr := deploymentsClient.Get(context.TODO(), "demo-deployment", metav1.GetOptions{})if getErr != nil {panic(fmt.Errorf("Failed to get latest version of Deployment: %v", getErr))}result.Spec.Replicas = int32Ptr(1) // reduce replica countresult.Spec.Template.Spec.Containers[0].Image = "nginx:1.13" // change nginx version_, updateErr := deploymentsClient.Update(context.TODO(), result, metav1.UpdateOptions{})return updateErr})if retryErr != nil {panic(fmt.Errorf("Update failed: %v", retryErr))}fmt.Println("Updated deployment...")// List Deploymentsprompt()fmt.Printf("Listing deployments in namespace %q:\n", apiv1.NamespaceDefault)list, err := deploymentsClient.List(context.TODO(), metav1.ListOptions{})if err != nil {panic(err)}for _, d := range list.Items {fmt.Printf(" * %s (%d replicas)\n", d.Name, *d.Spec.Replicas)}// Delete Deploymentprompt()fmt.Println("Deleting deployment...")deletePolicy := metav1.DeletePropagationForegroundif err := deploymentsClient.Delete(context.TODO(), "demo-deployment", metav1.DeleteOptions{PropagationPolicy: &deletePolicy,}); err != nil {panic(err)}fmt.Println("Deleted deployment.")
}func prompt() {fmt.Printf("-> Press Return key to continue.")scanner := bufio.NewScanner(os.Stdin)for scanner.Scan() {break}if err := scanner.Err(); err != nil {panic(err)}fmt.Println()
}func int32Ptr(i int32) *int32 { return &i }
4.deployment资源CRUD - dynamic方式
使用typed方式创建资源虽然方便,但是需要事先使用client-gen生成且会与某个特定版本耦合,不够通用,所以我们可以使用 unstructured.Unstructured 来表示集群中任意对象。
client, err := dynamic.NewForConfig(config)if err != nil {panic(err)}deploymentRes := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}deployment := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "apps/v1","kind": "Deployment","metadata": map[string]interface{}{"name": "demo-deployment",},"spec": map[string]interface{}{"replicas": 2,"selector": map[string]interface{}{"matchLabels": map[string]interface{}{"app": "demo",},},"template": map[string]interface{}{"metadata": map[string]interface{}{"labels": map[string]interface{}{"app": "demo",},},"spec": map[string]interface{}{"containers": []map[string]interface{}{{"name": "web","image": "nginx:1.12","ports": []map[string]interface{}{{"name": "http","protocol": "TCP","containerPort": 80,},},},},},},},},}// Create Deploymentfmt.Println("Creating deployment...")result, err := client.Resource(deploymentRes).Namespace(apiv1.NamespaceDefault).Create(context.TODO(), deployment, metav1.CreateOptions{})if err != nil {panic(err)}fmt.Printf("Created deployment %q.\n", result.GetName())
5. discoveryClient
3-4小节我们学习了 typed client 和 dynamic client ,这两个client使用来操作集群中资源,比如创建 deployment,删除pod。discovery client 不同,他是用来发现服务端支持的组、版本、资源类型:
// Package discovery provides ways to discover server-supported
// API groups, versions and resources.
我们可以查询服务端支持的组与版本、某个组的资源列表
func main() {var kubeconfig *stringif 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 kubeconfigconfig, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)if err != nil {panic(err.Error())}// *查询服务端支持的组列表discoveryClient := discovery.NewDiscoveryClientForConfigOrDie(config)gs, err := discoveryClient.ServerGroups()if err != nil {panic(err.Error())}spew.Dump(gs.Groups)// *查询core/v1 组下的资源列表rs, err := discoveryClient.ServerResourcesForGroupVersion("v1")if err != nil {panic(err.Error())}for _, r := range rs.APIResources {fmt.Println(r.Name)}}// 输出支持的组列表
/*
(v1.APIGroup) &APIGroup{Name:apps,Versions:[]GroupVersionForDiscovery{GroupVersionForDiscovery{GroupVersion:apps/v1,Version:v1,},},PreferredVersion:GroupVersionForDiscovery{GroupVersion:apps/v1,Version:v1,},ServerAddressByClientCIDRs:[]ServerAddressByClientCIDR{},},
(v1.APIGroup) &APIGroup{Name:events.k8s.io,Versions:[]GroupVersionForDiscovery{GroupVersionForDiscovery{GroupVersion:events.k8s.io/v1,Version:v1,},},PreferredVersion:GroupVersionForDiscovery{GroupVersion:events.k8s.io/v1,Version:v1,},ServerAddressByClientCIDRs:[]ServerAddressByClientCIDR{},},
...
*/// 输出core/v1组下的资源列表
/*
pods/status
podtemplates
replicationcontrollers
replicationcontrollers/scale
replicationcontrollers/status
resourcequotas
resourcequotas/status
secrets
serviceaccounts
...
*/
6. restClient
上面三种客户端都是基于 restClient 封装的,restClient 是 client-go 最基础的客户端,对http request 进行了封装,更加灵活,一般我们用不到。
// 根据配置信息构建restClient实例restClient, err := rest.RESTClientFor(config)if err!=nil {panic(err.Error())}// 保存pod结果的数据结构实例result := &corev1.PodList{}// 指定namespacenamespace := "kube-system"// 设置请求参数,然后发起请求// GET请求err = restClient.Get().// 指定namespace,参考path : /api/v1/namespaces/{namespace}/podsNamespace(namespace).// 查找多个pod,参考path : /api/v1/namespaces/{namespace}/podsResource("pods").// 指定大小限制和序列化工具VersionedParams(&metav1.ListOptions{Limit:100}, scheme.ParameterCodec).// 请求Do(context.TODO()).// 结果存入resultInto(result)
4、局限
使用 client 操作集群中资源会导致频繁的轮询,k8s client-go 包提供了更加高效的方式:informer 。
informer 提供了一种机制来监视 Kubernetes 集群内资源的变化并做出反应。它使开发人员能够接收有关各种 Kubernetes 对象(例如 Pod、服务、部署等)状态的实时更新。
Informer 将集群中的资源缓存在本地来减少频繁的API调用,提高性能并优化资源利用率。