百度360必应搜狗淘宝本站头条
python类型转换python计时python异常python进度条python吧
当前位置：网站首页 > 技术资源 > 正文

只想简单操作 Kubernetes 资源?用 Go 一招搞定，快试试 kom!

off999 2025-01-17 12:35 19 浏览 0 评论

当你在使用代码操作k8s时，有没有觉得总要写很多代码，才能获取到一个资源，并且要处理底层的集群连接、client等一系列的东西。那么有没有一种简单的办法，毕竟我只是想读取一下集群内的Pod或者什么其他资源而已啊。

为了解决这个问题，我们使用 kom
https://github.com/weibaohui/kom，这是一个k8s的操作类库，golang编写，使用简单。

在 Kubernetes 中，`Deployment` 是一种常见的控制器，负责管理一组副本的 Pod，确保这些 Pod 处于期望的状态。

通过 Go 客户端库和 `kom` 工具集，我们可以轻松地操作 Kubernetes 中的资源，包括创建、查询、更新、删除以及监听资源变更等。

本文将详细讲解如何使用 `kom` 对 `Deployment` 资源进行操作。

连接k8s集群

kom.Clusters().RegisterByPathWithID("/Users/kom/.kube/config", "docker-desktop")

#### 1. 定义一个 `Deployment` 对象

首先，创建一个 `Deployment` 对象，并设置相关参数。你可以通过 Go 语言中的结构体来定义它的元数据和规格。

var item v1.Deployment
item = v1.Deployment{
    ObjectMeta: metav1.ObjectMeta{
        Name:      "nginx",
        Namespace: "default",
    },
    Spec: v1.DeploymentSpec{
        Template: corev1.PodTemplateSpec{
            Spec: corev1.PodSpec{
                Containers: []corev1.Container{
                    {Name: "test", Image: "nginx:1.14.2"},
                },
            },
        },
    },
}

2. 创建资源

通过 `kom` 库，你可以轻松地在 Kubernetes 集群中创建一个 `Deployment` 资源。

err := kom.DefaultCluster().Resource(&item).Create(&item).Error
if err != nil {
    fmt.Printf("Error creating deployment: %v", err)
}

3. 查询资源

你可以通过 `Get` 方法获取特定的 `Deployment` 资源，例如查询 `default` 命名空间中的 `nginx` 部署：

err := kom.DefaultCluster().Resource(&item).Namespace("default").Name("nginx").Get(&item).Error
if err != nil {
    fmt.Printf("Error retrieving deployment: %v", err)
}

4. 查询资源列表

你还可以查询命名空间中的所有 `Deployment` 资源，或者通过标签进行过滤。例如，查询 `default` 命名空间下所有标签为 `app=nginx` 的 `Deployment`：

err := kom.DefaultCluster().Resource(&item).Namespace("default").List(&items, metav1.ListOptions{LabelSelector: "app=nginx"}).Error
if err != nil {
    fmt.Printf("Error listing deployments: %v", err)
}

通过字段选择器，你也可以按照指定的字段来筛选资源。例如，查询名为 `test-deploy` 的 `Deployment`：

err := kom.DefaultCluster().Resource(&item).Namespace("default").List(&items, metav1.ListOptions{FieldSelector: "metadata.name=test-deploy"}).Error
if err != nil {
    fmt.Printf("Error listing deployments: %v", err)
}

5. 更新资源

更新资源的常见场景之一是修改 `Deployment` 的注解。假设我们需要为名为 `nginx` 的 `Deployment` 增加一个注解：

err := kom.DefaultCluster().Resource(&item).Namespace("default").Name("nginx").Get(&item).Error
if err != nil {
    fmt.Printf("Error retrieving deployment for update: %v", err)
}
if item.Spec.Template.Annotations == nil {
    item.Spec.Template.Annotations = map[string]string{}
}
item.Spec.Template.Annotations["kom.kubernetes.io/restartedAt"] = time.Now().Format(time.RFC3339)
err = kom.DefaultCluster().Resource(&item).Update(&item).Error
if err != nil {
    fmt.Printf("Error updating deployment: %v", err)
}

6. 使用 Patch 更新资源

有时，直接修改资源较为繁琐，这时可以使用 `Patch` 方法来部分更新资源。例如，我们想给 `nginx` 部署添加一个标签，并设置副本数为 5：

patchData := `{
    "spec": {
        "replicas": 5
    },
    "metadata": {
        "labels": {
            "new-label": "new-value"
        }
    }
}`
err := kom.DefaultCluster().Resource(&item).Patch(&item, types.MergePatchType, patchData).Error
if err != nil {
    fmt.Printf("Error patching deployment: %v", err)
}

7. 删除资源

删除 `Deployment` 资源同样简单，只需要调用 `Delete` 方法：

err := kom.DefaultCluster().Resource(&item).Namespace("default").Name("nginx").Delete().Error
if err != nil {
    fmt.Printf("Error deleting deployment: %v", err)
}

8. 获取通用类型资源

对于 Kubernetes 中的内置类型或自定义资源定义 (CRD)，你可以通过指定资源的 `GroupVersionKind` (GVK) 来获取资源。例如，获取 `Event` 类型的资源：

var list []corev1.Event
err := kom.DefaultCluster().GVK("events.k8s.io", "v1", "Event").Namespace("default").List(&list).Error
if err != nil {
    fmt.Printf("Error listing events: %v", err)
}

9. Watch 资源变更

`Watch` 方法可以用来监听资源的实时变更，适用于处理需要实时反应的场景。例如，监听 `default` 命名空间下 `Pod` 资源的变动：

var watcher watch.Interface
var pod corev1.Pod
err := kom.DefaultCluster().Resource(&pod).Namespace("default").Watch(&watcher).Error
if err != nil {
    fmt.Printf("Error creating watcher: %v", err)
    return
}
go func() {
    defer watcher.Stop()

    for event := range watcher.ResultChan() {
        err := kom.DefaultCluster().Tools().ConvertRuntimeObjectToTypedObject(event.Object, &pod)
        if err != nil {
            fmt.Printf("Error converting event object: %v", err)
            return
        }
        switch event.Type {
        case watch.Added:
            fmt.Printf("Added Pod [ %s/%s ]\n", pod.Namespace, pod.Name)
        case watch.Modified:
            fmt.Printf("Modified Pod [ %s/%s ]\n", pod.Namespace, pod.Name)
        case watch.Deleted:
            fmt.Printf("Deleted Pod [ %s/%s ]\n", pod.Namespace, pod.Name)
        }
    }
}()

对比

我们来看下，如果不使用kom，我们如何来操作Deployment资源

package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

func main() {
	// 加载 kubeconfig
	kubeconfig := flag.String("kubeconfig", "/path/to/your/kubeconfig", "absolute path to the kubeconfig file")
	flag.Parse()
	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		log.Fatalf("Failed to build kubeconfig: %v", err)
	}

	// 创建 Kubernetes 客户端
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Fatalf("Failed to create clientset: %v", err)
	}

	// Namespace 定义
	namespace := "default"

	// 示例操作
	ctx := context.TODO()

	// 创建 Deployment
	createDeployment(ctx, clientset, namespace)

	// 获取 Deployment
	getDeployment(ctx, clientset, namespace, "example-deployment")

	// 更新 Deployment
	updateDeployment(ctx, clientset, namespace, "example-deployment")

	// Patch Deployment
	patchDeployment(ctx, clientset, namespace, "example-deployment")

	// 列出 Deployments
	listDeployments(ctx, clientset, namespace)
}

func createDeployment(ctx context.Context, clientset *kubernetes.Clientset, namespace string) {
	deployment := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name: "example-deployment",
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: int32Ptr(3),
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					"app": "example",
				},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						"app": "example",
					},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:  "nginx",
							Image: "nginx:1.21",
							Ports: []corev1.ContainerPort{
								{
									ContainerPort: 80,
								},
							},
						},
					},
				},
			},
		},
	}

	_, err := clientset.AppsV1().Deployments(namespace).Create(ctx, deployment, metav1.CreateOptions{})
	if err != nil {
		log.Fatalf("Failed to create deployment: %v", err)
	}
	fmt.Println("Deployment created successfully!")
}

func getDeployment(ctx context.Context, clientset *kubernetes.Clientset, namespace, name string) {
	deployment, err := clientset.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		log.Fatalf("Failed to get deployment: %v", err)
	}
	fmt.Printf("Deployment retrieved: %s with %d replicas\n", deployment.Name, *deployment.Spec.Replicas)
}

func updateDeployment(ctx context.Context, clientset *kubernetes.Clientset, namespace, name string) {
	deployment, err := clientset.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		log.Fatalf("Failed to get deployment for update: %v", err)
	}

	// 修改副本数
	newReplicas := int32(5)
	deployment.Spec.Replicas = &newReplicas

	_, err = clientset.AppsV1().Deployments(namespace).Update(ctx, deployment, metav1.UpdateOptions{})
	if err != nil {
		log.Fatalf("Failed to update deployment: %v", err)
	}
	fmt.Println("Deployment updated successfully!")
}

func patchDeployment(ctx context.Context, clientset *kubernetes.Clientset, namespace, name string) {
	patchData := map[string]interface{}{
		"spec": map[string]interface{}{
			"replicas": 2,
		},
	}
	patchBytes, _ := json.Marshal(patchData)

	_, err := clientset.AppsV1().Deployments(namespace).Patch(ctx, name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{})
	if err != nil {
		log.Fatalf("Failed to patch deployment: %v", err)
	}
	fmt.Println("Deployment patched successfully!")
}

func listDeployments(ctx context.Context, clientset *kubernetes.Clientset, namespace string) {
	deployments, err := clientset.AppsV1().Deployments(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		log.Fatalf("Failed to list deployments: %v", err)
	}
	fmt.Println("Deployments in namespace:", namespace)
	for _, d := range deployments.Items {
		fmt.Printf("- %s (replicas: %d)\n", d.Name, *d.Spec.Replicas)
	}
}

func int32Ptr(i int32) *int32 { return &i }

那么这么写的问题在于，如果你需要操作多个 Kubernetes 资源类型（如 Deployment、Ingress、Service 等），你可能会发现每个资源类型的操作都需要写一套相似的代码，主要是：

创建资源：每个资源类型有自己特定的 Spec 和 Meta 信息。

获取资源：每个资源类型有自己的获取方法。

更新、删除资源：这些操作的代码也是基于资源类型的差异化进行的。

然而，使用kom，则可以避免处理这些具体类型的资源直接的差异信息。

总结

通过 `kom` 库，我们可以方便地在 Go 中对 Kubernetes 资源进行操作，包括创建、查询、更新、删除以及监听资源变更等。通过这些操作，开发者可以更高效地与 Kubernetes 集群交互，管理部署过程。

引用

https://github.com/weibaohui/k8m

https://github.com/weibaohui/kom

相关推荐

安装python语言,运行你的第一行代码

#01安装Python访问Python官方(https://www.python.org/)，下载并安装最新版本的Python。确保安装过程中勾选“Addpython.exetoPAT...

Python推导式家族深度解析:字典/集合/生成器的艺术

一、为什么需要其他推导式？当你在处理数据时：o需要快速去重→集合推导式o要建立键值映射→字典推导式o处理海量数据→生成器表达式这些场景是列表推导式无法完美解决的，就像工具箱需要不同工...

别再用循环创建字典了!Python推导式让你的代码起飞

当同事还在用for循环吭哧吭哧创建字典时，我早已用推导式完成3个需求了！这个被90%新手忽视的语法，今天让你彻底掌握字典推导式的4大高阶玩法，文末彩蛋教你用1行代码搞定复杂数据转换！基础语法拆解#传...

什么是Python中的生成器推导式?（python生成器的好处）

编程派微信号：codingpy本文作者为NedBatchelder，是一名资深Python工程师，目前就职于在线教育网站Edx。文中蓝色下划线部分可“阅读原文”后点击。Python中有一种紧凑的语法...

Python 列表转换为字符串:实用指南

为什么在Python中将列表转换为字符串？Python列表非常灵活，但它们并非在所有地方都适用。有时你需要以人类可读的格式呈现数据——比如在UI中显示标签或将项目保存到CSV文件。可能还...

生成器表达式和列表推导式（生成器表达式的计算结果）

迭代器的输出有两个很常见的使用方式，1)对每一个元素执行操作，2)选择一个符合条件的元素子集。比如，给定一个字符串列表，你可能想去掉每个字符串尾部的空白字符，或是选出所有包含给定子串的字符串。列表...

python学习——038python中for循环VS列表推导式

在Python中，for循环和列表推导式（ListComprehension）都可以用于创建和处理列表，但它们的语法、性能和适用场景有所不同。以下是两者的详细对比：1.语法结构for循环使用...

python中列表推导式怎么用?（列表 python）

这个问题，我们不妨用近期很火的ChatGPT来试试，来看看人工智能是如何解答的？在Python中，列表解析是一种简洁的方法，用于生成列表。它是一种快速，简洁的方法，可以在一行代码中生成列表，而不需...

Python列表推导式:让你的代码优雅如诗!

每次写for循环都要三四行代码？处理数据时总被嵌套结构绕晕？学会列表推导式，一行代码就能让代码简洁十倍！今天带你解锁这个Python程序员装（偷）逼（懒）神器！一、为什么你需要列表推导式？代码...

python学习——038如何将for循环改写成列表推导式

在Python里，列表推导式是一种能够简洁生成列表的表达式，可用于替换普通的for循环。下面是列表推导式的基本语法和常见应用场景。基本语法result=[]foriteminite...

太牛了!Python 列表推导式，超级总结!这分析总结也太到位了!

Python列表推导式，超级总结！一、基本概念列表推导式是Python中创建列表的一种简洁语法，它允许你在一行代码内生成列表，替代传统的for循环方式。其核心思想是**"对可迭代对...

25-2-Python网络编程-TCP 编程示例

2-TCP编程示例应用程序通常通过“套接字”（socket）向网络发出请求或者应答网络请求，使主机间或者一台计算机上的进程间可以通信。Python语言提供了两种访问网络服务的功能。其中低级别的网络服...

python编程的基础与进阶(周兴富)（python编程基础视频）

前不久我发文：《懂了，if__name=='__main__'》。想不到的是，这个被朋友称之为“读晕了”的文章，其收藏量数百，有效阅读量竟然过万。所谓“有效阅读量”，就是读到尾部才退...

Python 闭包:深入理解函数式编程的核心概念

一、简介在Python编程领域，闭包（Closure）是一个既基础又强大的概念，它不仅是装饰器、回调函数等高级特性的实现基础，更是函数式编程思想的重要体现。理解闭包的工作原理，能够帮助开发者编写出...

Python小白逆袭!7天吃透PyQt6，独立开发超酷桌面应用

PythonGUI编程：PyQt6从入门到实战的全面指南在Python的庞大生态系统中，PyQt6作为一款强大的GUI（GraphicalUserInterface，图形用户界面）编程框架，为开...

取消回复欢迎 你 发表评论:

一周热门
最近发表
标签列表
Copyright Your WebSite.Some Rights Reserved.