Agent + K8s 初步开发#

1
// 创建Clientset客户端
2
func CreateClientset() (*kubernetes.Clientset, error) {
3
    // 1. 先加载配置（后面会详细介绍）
4
    config, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
5
    if err != nil {
6
        return nil, err
7
    }
8

9
    // 2. 基于配置创建Clientset
10
    clientset, err := kubernetes.NewForConfig(config)
11
    if err != nil {
12
        return nil, err
13
    }
14

15
    return clientset, nil
16
}
17

18
// 使用示例：列出所有运行中的Pod
19
func ListRunningPods(clientset *kubernetes.Clientset) {
20
    // CoreV1() - 返回CoreV1Interface接口
21
    // 功能：访问核心API组的v1版本资源（Pod、Service、ConfigMap等）
22
    // 其他组：AppsV1() - Deployment、StatefulSet
23
    //       BatchV1() - Job、CronJob
24

25
    // Pods("default") - 返回PodInterface接口
26
    // 参数：namespace字符串，指定要操作的命名空间
27
    // 功能：获取该命名空间下Pod资源的操作接口
28

29
    // List() - 执行列表查询
30
    // 参数：
31
    //   - context.TODO()：上下文，可控制超时、取消等
32
    //   - metav1.ListOptions：查询选项
33
    //     - FieldSelector：字段选择器，按资源字段过滤
34
    //     - LabelSelector：标签选择器，按标签过滤
35
    //     - Limit：限制返回数量
36
    //     - Continue：分页标记
37
    pods, err := clientset.
38
        CoreV1().
39
        Pods("default").
40
        List(context.TODO(), metav1.ListOptions{
41
            FieldSelector: "status.phase=Running",  // 只获取运行中的Pod
42
        })
43

44
    if err != nil {
45
        log.Fatal(err)
46
    }
47

48
    // pods.Items - []v1.Pod类型，强类型数组
49
    // 每个Pod对象都有完整的类型定义，IDE可以自动提示
50
    fmt.Println("运行中的Pod列表：")
51
    for _, pod := range pods.Items {
52
        fmt.Printf("- %s (创建时间: %s)\n",
53
            pod.Name,                                          // 直接访问字段
54
            pod.CreationTimestamp.Format("2006-01-02 15:04:05"))
55
    }
56
}

1
// 创建Dynamic客户端
2
func CreateDynamicClient() (dynamic.Interface, error) {
3
    // 1. 加载配置
4
    config, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
5
    if err != nil {
6
        return nil, err
7
    }
8

9
    // 2. 创建动态客户端
10
    dynamicClient, err := dynamic.NewForConfig(config)
11
    if err != nil {
12
        return nil, err
13
    }
14

15
    return dynamicClient, nil
16
}
17

18
// 使用示例：操作自定义资源
19
func OperateCustomResource(dynamicClient dynamic.Interface) {
20
    // 定义GVR - 这是动态客户端的核心概念
21
    // GVR (GroupVersionResource) 唯一标识一种资源类型
22
    gvr := schema.GroupVersionResource{
23
        Group:    "mycompany.com",  // API组，CRD通常使用域名
24
        Version:  "v1",             // API版本
25
        Resource: "myapps",         // 资源名称（必须是复数）
26
    }
27

28
    // 创建一个自定义资源对象
29
    // unstructured.Unstructured - 非结构化数据
30
    // 本质是 map[string]interface{} 的封装
31
    // 可以表示任意K8s资源，无需预定义Go结构体
32
    myApp := &unstructured.Unstructured{
33
        Object: map[string]interface{}{
34
            "apiVersion": "mycompany.com/v1",  // 必需：API版本
35
            "kind":       "MyApp",              // 必需：资源种类
36
            "metadata": map[string]interface{}{ // 必需：元数据
37
                "name":      "example-app",
38
                "namespace": "default",
39
            },
40
            "spec": map[string]interface{}{    // 自定义字段
41
                "replicas": 3,
42
                "image":    "nginx:latest",
43
            },
44
        },
45
    }
46

47
    // Resource(gvr) - 指定要操作的资源类型
48
    // Namespace("default") - 指定命名空间
49
    // Create() - 执行创建操作
50
    result, err := dynamicClient.
51
        Resource(gvr).
52
        Namespace("default").
53
        Create(context.TODO(), myApp, metav1.CreateOptions{})
54

55
    if err != nil {
56
        log.Fatal(err)
57
    }
58

59
    // result也是 *unstructured.Unstructured 类型
60
    // 使用 GetXXX 方法访问字段
61
    fmt.Printf("创建成功: %s\n", result.GetName())
62
}

1
// 创建RESTClient
2
func CreateRESTClient() (*rest.RESTClient, error) {
3
    // 1. 加载基础配置
4
    config, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
5
    if err != nil {
6
        return nil, err
7
    }
8

9
    // 2. RESTClient需要额外配置
10
    // 必须指定API路径和Group/Version信息
11
    config.APIPath = "/api"  // API路径前缀
12
    config.GroupVersion = &corev1.SchemeGroupVersion  // 指定要访问的API组版本
13
    config.NegotiatedSerializer = scheme.Codecs.WithoutConversion()  // 序列化器
14

15
    // 3. 创建RESTClient
16
    restClient, err := rest.RESTClientFor(config)
17
    if err != nil {
18
        return nil, err
19
    }
20

21
    return restClient, nil
22
}
23

24
// 使用示例：获取Pod日志（这是Clientset做不到的）
25
func GetPodLogs(restClient *rest.RESTClient) {
26
    // Get() - 创建GET请求构建器
27
    // AbsPath() - 使用绝对路径（从根/开始）
28
    // 路径格式：/api/v1/namespaces/{namespace}/pods/{name}/log
29
    req := restClient.
30
        Get().
31
        AbsPath("/api/v1/namespaces/default/pods/my-pod/log").
32
        Param("follow", "true").       // 添加查询参数：实时跟踪
33
        Param("container", "nginx")    // 指定容器
34

35
    // Stream() - 获取响应流（适合日志等流式数据）
36
    // 也可以用 Do() 获取一次性响应
37
    stream, err := req.Stream(context.TODO())
38
    if err != nil {
39
        log.Fatal(err)
40
    }
41
    defer stream.Close()
42

43
    // 读取日志流
44
    buf := make([]byte, 1024)
45
    for {
46
        n, err := stream.Read(buf)
47
        if err != nil {
48
            break
49
        }
50
        fmt.Print(string(buf[:n]))
51
    }
52
}

1
// POST请求 - 创建资源
2
restClient.Post().AbsPath("/api/v1/namespaces/default/pods").Body(podObj).Do(ctx)
3

4
// DELETE请求 - 删除资源
5
restClient.Delete().AbsPath("/api/v1/namespaces/default/pods/my-pod").Do(ctx)
6

7
// PATCH请求 - 部分更新
8
restClient.Patch(types.MergePatchType).AbsPath("...").Body(patchData).Do(ctx)

1
// 创建Discovery客户端
2
func CreateDiscoveryClient() (*discovery.DiscoveryClient, error) {
3
    config, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
4
    if err != nil {
5
        return nil, err
6
    }
7

8
    // 创建Discovery客户端
9
    discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
10
    if err != nil {
11
        return nil, err
12
    }
13

14
    return discoveryClient, nil
15
}
16

17
// 使用示例：发现集群支持的所有资源
18
func DiscoverClusterResources(discoveryClient *discovery.DiscoveryClient) {
19
    // ServerGroups() - 获取所有API组信息
20
    // 返回 *metav1.APIGroupList
21
    apiGroups, err := discoveryClient.ServerGroups()
22
    if err != nil {
23
        log.Fatal(err)
24
    }
25

26
    fmt.Println("集群支持的API组：")
27
    for _, group := range apiGroups.Groups {
28
        // 空字符串表示核心组（v1）
29
        if group.Name == "" {
30
            fmt.Println("\n核心组 (core):")
31
        } else {
32
            fmt.Printf("\n组名: %s\n", group.Name)
33
        }
34

35
        fmt.Println("  支持的版本:")
36
        for _, version := range group.Versions {
37
            // GroupVersion格式: "apps/v1", "batch/v1"
38
            fmt.Printf("    - %s\n", version.Version)
39
        }
40

41
        // PreferredVersion - 该组的首选版本
42
        fmt.Printf("  首选版本: %s\n", group.PreferredVersion.Version)
43
    }
44

45
    // ServerResourcesForGroupVersion() - 获取特定组版本的所有资源
46
    resources, err := discoveryClient.ServerResourcesForGroupVersion("apps/v1")
47
    if err == nil {
48
        fmt.Println("\napps/v1 组的资源：")
49
        for _, r := range resources.APIResources {
50
            // Name: 资源名称（复数）
51
            // Kind: 资源种类
52
            // Namespaced: 是否是命名空间级资源
53
            // Verbs: 支持的操作 ["create", "delete", "get", "list", "patch", "update", "watch"]
54
            fmt.Printf("  - %s (Kind: %s, Namespaced: %v)\n",
55
                r.Name, r.Kind, r.Namespaced)
56
        }
57
    }
58

59
    // ServerVersion() - 获取K8s版本信息
60
    version, _ := discoveryClient.ServerVersion()
61
    fmt.Printf("\nKubernetes版本: %s\n", version.GitVersion)
62
}

1
// GVR示例
2
gvr := schema.GroupVersionResource{
3
    Group:    "apps",        // API组
4
    Version:  "v1",         // 版本
5
    Resource: "deployments", // 资源（复数形式）
6
}
7
// 对应的API路径：/apis/apps/v1/deployments
8
// 如果Group为空（核心组）：/api/v1/pods

1
// GVK示例
2
gvk := schema.GroupVersionKind{
3
    Group:   "apps",       // API组
4
    Version: "v1",         // 版本
5
    Kind:    "Deployment", // 种类（单数形式，首字母大写）
6
}
7
// 对应YAML中：
8
// apiVersion: apps/v1
9
// kind: Deployment

1
// 区别对照表
2
// ┌─────────────┬──────────────────┬─────────────────┐
3
// │   属性       │      GVR         │      GVK        │
4
// ├─────────────┼──────────────────┼─────────────────┤
5
// │ 用途        │ API路径/URL构建   │ 对象类型标识     │
6
// │ 资源名称     │ 复数(deployments) │ 单数(Deployment) │
7
// │ 使用场景     │ 客户端API调用     │ YAML/JSON定义    │
8
// │ 示例        │ pods、services   │ Pod、Service    │
9
// └─────────────┴──────────────────┴─────────────────┘
10

11
// 转换示例
12
func ShowGVRAndGVK() {
13
    // 用户视角：操作Deployment
14
    // GVK - 用户在YAML中写的
15
    gvk := schema.GroupVersionKind{
16
        Group:   "apps",
17
        Version: "v1",
18
        Kind:    "Deployment",
19
    }
20

21
    // GVR - 实际API调用使用的
22
    gvr := schema.GroupVersionResource{
23
        Group:    "apps",
24
        Version:  "v1",
25
        Resource: "deployments",  // 注意：复数
26
    }
27

28
    // RESTMapper的作用就是在两者之间转换
29
    // Kind -> Resource: Deployment -> deployments
30
    // Resource -> Kind: pods -> Pod
31
}

1
// 初始化RESTMapper
2
func InitRestMapper(clientSet *kubernetes.Clientset) meta.RESTMapper {
3
    // 这里看起来传入的是Clientset，但实际使用的是其Discovery()方法
4
    // clientSet.Discovery() 返回 discovery.DiscoveryInterface
5
    // 原因：Clientset内部包含了Discovery客户端
6

7
    // GetAPIGroupResources使用Discovery客户端来：
8
    // 1. 调用 /api 和 /apis 端点
9
    // 2. 获取所有API组的信息
10
    // 3. 获取每个组支持的资源类型
11
    // 4. 构建资源映射关系
12
    gr, err := restmapper.GetAPIGroupResources(clientSet.Discovery())
13
    if err != nil {
14
        panic(err)
15
    }
16

17
    // 创建映射器，内部构建了多个映射表：
18
    // - Resource -> Kind (pods -> Pod)
19
    // - Kind -> Resource (Service -> services)
20
    // - 资源的作用域信息（命名空间级/集群级）
21
    // - 资源的首选版本信息
22
    mapper := restmapper.NewDiscoveryRESTMapper(gr)
23
    return mapper
24
}

1
// RESTMapper简洁工作流程：
2
// 1. 初始化阶段（启动时执行一次）
3
//    Discovery客户端 -> 获取所有API信息 -> 构建映射表
4
//
5
// 2. 运行时映射（处理用户请求）
6
//    用户输入 -> 解析 -> 查询映射表 -> 返回完整资源信息
7

8
func MappingWorkflow() {
9
    // 步骤1: 用户输入各种格式
10
    userInputs := []string{
11
        "pod",              // 简单资源名
12
        "pods",             // 复数形式
13
        "deployment",       // 简写
14
        "deploy",          // 更短的简写
15
        "pods.v1",         // 带版本
16
        "deployments.apps", // 带组名
17
    }
18

19
    // 步骤2: RESTMapper处理流程
20
    // parseInput -> 解析输入格式
21
    // lookupResource -> 查找资源定义
22
    // resolveVersion -> 确定最佳API版本
23
    // buildMapping -> 构建完整映射信息
24

25
    // 步骤3: 返回RESTMapping对象，包含：
26
    // - GVR: 用于API调用
27
    // - GVK: 资源类型信息
28
    // - Scope: 命名空间级还是集群级
29
    // - 其他元信息
30
}

1
// 智能映射函数 - 处理各种用户输入
2
func MappingFor(userInput string, restMapper meta.RESTMapper) (*meta.RESTMapping, error) {
3
    // 尝试解析为Resource（资源名称）
4
    // ParseResourceArg处理格式：pods、pods.v1、deployments.apps.v1
5
    fullySpecifiedGVR, groupResource := schema.ParseResourceArg(userInput)
6

7
    // 尝试解析为Kind（种类名称）
8
    // ParseKindArg处理格式：Pod、Pod.v1、Deployment.apps
9
    fullySpecifiedGVK, groupKind := schema.ParseKindArg(userInput)
10

11
    var mapping *meta.RESTMapping
12
    var err error
13

14
    // 策略1：如果是完整的GVR，先转换为GVK，再获取映射
15
    if fullySpecifiedGVR != nil {
16
        gvk, _ := restMapper.KindFor(*fullySpecifiedGVR)
17
        if !gvk.Empty() {
18
            mapping, err = restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
19
        }
20
    }
21

22
    // 策略2：如果是GVK格式，直接获取映射
23
    if mapping == nil && fullySpecifiedGVK != nil {
24
        mapping, err = restMapper.RESTMapping(
25
            fullySpecifiedGVK.GroupKind(),
26
            fullySpecifiedGVK.Version,
27
        )
28
    }
29

30
    // 策略3：只有组和种类，使用首选版本
31
    if mapping == nil && !groupKind.Empty() {
32
        mapping, err = restMapper.RESTMapping(groupKind)
33
    }
34

35
    return mapping, err
36
}

1
// 用户输入 -> K8s资源
2
"pod"        -> GVR{Group:"", Version:"v1", Resource:"pods"}
3
"deploy"     -> GVR{Group:"apps", Version:"v1", Resource:"deployments"}
4
"svc"        -> GVR{Group:"", Version:"v1", Resource:"services"}
5
"node"       -> GVR{Group:"", Version:"v1", Resource:"nodes"} + Scope:Cluster
6
"ns"         -> GVR{Group:"", Version:"v1", Resource:"namespaces"} + Scope:Cluster

1
package services
2

3
import (
4
    "context"
5
    "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
6
    "k8s.io/apimachinery/pkg/runtime/schema"
7
    "k8s.io/client-go/dynamic"
8
)
9

10
type ResourceService struct {
11
    dynamicClient dynamic.Interface
12
    restMapper    meta.RESTMapper
13
    informerFactory informers.SharedInformerFactory
14
}
15

16
// 创建资源 - 支持YAML输入
17
func (s *ResourceService) CreateResource(
18
    resourceType string,  // 如 "deployment", "service"
19
    namespace string,
20
    yamlContent string,
21
) error {
22
    // 1. 解析YAML
23
    obj := &unstructured.Unstructured{}
24
    _, _, err := scheme.Codecs.UniversalDeserializer().
25
        Decode([]byte(yamlContent), nil, obj)
26
    if err != nil {
27
        return fmt.Errorf("YAML解析失败: %v", err)
28
    }
29

30
    // 2. 获取资源映射
31
    mapping, err := MappingFor(resourceType, s.restMapper)
32
    if err != nil {
33
        return fmt.Errorf("未知的资源类型: %s", resourceType)
34
    }
35

36
    // 3. 判断资源作用域
37
    var resourceInterface dynamic.ResourceInterface
38
    if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
39
        // 命名空间级资源
40
        resourceInterface = s.dynamicClient.
41
            Resource(mapping.Resource).
42
            Namespace(namespace)
43
    } else {
44
        // 集群级资源
45
        resourceInterface = s.dynamicClient.
46
            Resource(mapping.Resource)
47
    }
48

49
    // 4. 创建资源
50
    _, err = resourceInterface.Create(
51
        context.Background(),
52
        obj,
53
        metav1.CreateOptions{},
54
    )
55

56
    return err
57
}
58

59
// 列出资源 - 从缓存读取
60
func (s *ResourceService) ListResources(
61
    resourceType string,
62
    namespace string,
63
) ([]unstructured.Unstructured, error) {
64
    // 获取资源映射
65
    mapping, err := MappingFor(resourceType, s.restMapper)
66
    if err != nil {
67
        return nil, err
68
    }
69

70
    // 从Informer缓存获取
71
    informer, err := s.informerFactory.ForResource(mapping.Resource)
72
    if err != nil {
73
        return nil, err
74
    }
75

76
    // 列出资源
77
    var items []unstructured.Unstructured
78
    if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
79
        objs, _ := informer.Lister().ByNamespace(namespace).List(labels.Everything())
80
        for _, obj := range objs {
81
            items = append(items, *obj.(*unstructured.Unstructured))
82
        }
83
    } else {
84
        objs, _ := informer.Lister().List(labels.Everything())
85
        for _, obj := range objs {
86
            items = append(items, *obj.(*unstructured.Unstructured))
87
        }
88
    }
89

90
    return items, nil
91
}
92

93
// 删除资源
94
func (s *ResourceService) DeleteResource(
95
    resourceType string,
96
    namespace string,
97
    name string,
98
) error {
99
    mapping, err := MappingFor(resourceType, s.restMapper)
100
    if err != nil {
101
        return err
102
    }
103

104
    var resourceInterface dynamic.ResourceInterface
105
    if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
106
        resourceInterface = s.dynamicClient.
107
            Resource(mapping.Resource).
108
            Namespace(namespace)
109
    } else {
110
        resourceInterface = s.dynamicClient.
111
            Resource(mapping.Resource)
112
    }
113

114
    return resourceInterface.Delete(
115
        context.Background(),
116
        name,
117
        metav1.DeleteOptions{},
118
    )
119
}

1
package controllers
2

3
import (
4
    "github.com/gin-gonic/gin"
5
    "your-project/services"
6
)
7

8
type ResourceController struct {
9
    service *services.ResourceService
10
}
11

12
// POST /:resource - 创建资源
13
func (c *ResourceController) Create() gin.HandlerFunc {
14
    return func(ctx *gin.Context) {
15
        resourceType := ctx.Param("resource")
16
        namespace := ctx.DefaultQuery("namespace", "default")
17

18
        var body struct {
19
            Yaml string `json:"yaml" binding:"required"`
20
        }
21

22
        if err := ctx.ShouldBindJSON(&body); err != nil {
23
            ctx.JSON(400, gin.H{
24
                "error": "请提供有效的YAML内容",
25
            })
26
            return
27
        }
28

29
        err := c.service.CreateResource(resourceType, namespace, body.Yaml)
30
        if err != nil {
31
            ctx.JSON(500, gin.H{
32
                "error": fmt.Sprintf("创建失败: %v", err),
33
            })
34
            return
35
        }
36

37
        ctx.JSON(201, gin.H{
38
            "message": fmt.Sprintf("%s 创建成功", resourceType),
39
        })
40
    }
41
}
42

43
// GET /:resource - 列出资源
44
func (c *ResourceController) List() gin.HandlerFunc {
45
    return func(ctx *gin.Context) {
46
        resourceType := ctx.Param("resource")
47
        namespace := ctx.DefaultQuery("namespace", "default")
48

49
        items, err := c.service.ListResources(resourceType, namespace)
50
        if err != nil {
51
            ctx.JSON(500, gin.H{
52
                "error": fmt.Sprintf("查询失败: %v", err),
53
            })
54
            return
55
        }
56

57
        // 简化输出，只返回名称列表
58
        var names []string
59
        for _, item := range items {
60
            names = append(names, item.GetName())
61
        }
62

63
        ctx.JSON(200, gin.H{
64
            "namespace": namespace,
65
            "resource":  resourceType,
66
            "count":     len(names),
67
            "items":     names,
68
        })
69
    }
70
}
71

72
// DELETE /:resource/:name - 删除资源
73
func (c *ResourceController) Delete() gin.HandlerFunc {
74
    return func(ctx *gin.Context) {
75
        resourceType := ctx.Param("resource")
76
        name := ctx.Param("name")
77
        namespace := ctx.DefaultQuery("namespace", "default")
78

79
        err := c.service.DeleteResource(resourceType, namespace, name)
80
        if err != nil {
81
            ctx.JSON(500, gin.H{
82
                "error": fmt.Sprintf("删除失败: %v", err),
83
            })
84
            return
85
        }
86

87
        ctx.JSON(200, gin.H{
88
            "message": fmt.Sprintf("%s/%s 删除成功", resourceType, name),
89
        })
90
    }
91
}

1
package main
2

3
import (
4
    "log"
5
    "github.com/gin-gonic/gin"
6
    "k8s.io/client-go/dynamic"
7
    "your-project/config"
8
    "your-project/services"
9
    "your-project/controllers"
10
)
11

12
func main() {
13
    // 1. 初始化K8s配置
14
    log.Println("正在初始化K8s配置...")
15
    k8sConfig := config.NewK8sConfig().InitRestConfig()
16
    if k8sConfig.Error() != nil {
17
        log.Fatal("配置加载失败:", k8sConfig.Error())
18
    }
19

20
    // 2. 创建客户端
21
    clientSet := k8sConfig.InitClientSet()
22
    dynamicClient, err := dynamic.NewForConfig(k8sConfig.Config)
23
    if err != nil {
24
        log.Fatal("创建动态客户端失败:", err)
25
    }
26

27
    // 3. 初始化RESTMapper
28
    log.Println("正在初始化资源映射器...")
29
    restMapper := InitRestMapper(clientSet)
30

31
    // 4. 初始化Informer
32
    log.Println("正在初始化缓存系统...")
33
    informerFactory := InitInformerFactory(clientSet)
34

35
    // 5. 创建服务层
36
    resourceService := &services.ResourceService{
37
        DynamicClient:   dynamicClient,
38
        RestMapper:      restMapper,
39
        InformerFactory: informerFactory,
40
    }
41

42
    // 6. 创建控制器
43
    resourceController := &controllers.ResourceController{
44
        Service: resourceService,
45
    }
46

47
    // 7. 设置路由
48
    router := gin.Default()
49

50
    // 资源操作路由
51
    router.POST("/:resource", resourceController.Create())
52
    router.GET("/:resource", resourceController.List())
53
    router.DELETE("/:resource/:name", resourceController.Delete())
54

55
    // 健康检查
56
    router.GET("/health", func(c *gin.Context) {
57
        c.JSON(200, gin.H{"status": "healthy"})
58
    })
59

60
    // 8. 启动服务
61
    log.Println("K8s Agent服务启动在 http://localhost:8080")
62
    if err := router.Run(":8080"); err != nil {
63
        log.Fatal("服务启动失败:", err)
64
    }
65
}