Client-go AIOps实战

# 1. 安装 Cobra CLI（仓库地址在注释里）
go install github.com/spf13/cobra-cli@latest

# 2. 初始化一个 Go Module
go mod init k8sco

# 3. 用 Cobra CLI 初始化项目，并指定 module path
cobra-cli init --module k8sco

k8sco/
├── cmd/
│   └── root.go      # 根命令定义
├── go.mod
└── main.go

// cmd/root.go
package cmd

import (
	"os"
	"github.com/spf13/cobra"
)

var rootCmd = &cobra.Command{
	Use:   "k8sco",
	Short: "一个结合 client-go 与 LLM 的 K8s AIOps 命令行工具",
	Long: `k8sco 是一个基于 Cobra 的 Kubernetes AIOps 命令行工具。
下面是一些用法示例：
  k8sco hello world      运行 hello 命令
  k8sco ask chatgpt      以对话方式操作 K8s 资源
  k8sco analyze events   获取集群事件并结合 LLM 做故障诊断`,
	// Version 一旦填上，命令行就自动获得 --version / version 能力
	Version: "v0.1.0",
}

func Execute() {
	err := rootCmd.Execute()
	if err != nil {
		os.Exit(1)
	}
}

go build -o k8sco .
./k8sco                # 输出根命令的 Long 描述（因为没有任何子命令，默认打印帮助）
./k8sco --version      # 直接输出 v0.1.0

cobra-cli add hello

// cmd/hello.go
package cmd

import (
	"fmt"
	"github.com/spf13/cobra"
)

var helloCmd = &cobra.Command{
	Use:   "hello",
	Short: "Hello 命令示例",
	Run: func(cmd *cobra.Command, args []string) {
		fmt.Println("hello k8sco")
	},
}

func init() {
	rootCmd.AddCommand(helloCmd)   // 把 hello 挂在根命令下面
}

cobra-cli add world --parent helloCmd

// cmd/world.go（核心结构）
var worldCmd = &cobra.Command{
	Use:   "world",
	Short: "World 子命令示例",
	Run: func(cmd *cobra.Command, args []string) {
		fmt.Println("world k8sco")
	},
}

func init() {
	helloCmd.AddCommand(worldCmd)   // 关键：挂到 helloCmd 下面，而不是 rootCmd
}

// 1. string 类型（不带缩写的长 flag）—— 如 --namespace
var ns string
rootCmd.PersistentFlags().StringVar(&ns, "namespace", "default", "目标命名空间")

// 2. string 类型，带短 flag 缩写 —— 如 -n / --namespace
var ns2 string
rootCmd.PersistentFlags().StringVarP(&ns2, "namespace", "n", "default", "目标命名空间")

// 3. 字符串切片（可传多个）—— 如 --node a --node b，存进 []string
var nodes []string
rootCmd.PersistentFlags().StringSliceVar(&nodes, "node", nil, "节点列表（可多次指定）")

// 4. 布尔、整数、浮点等类型的数据也有对应方法
var verbose bool
var replicas int
var ratio float64
rootCmd.PersistentFlags().BoolVar(&verbose, "verbose", false, "是否输出详细日志")
rootCmd.PersistentFlags().IntVar(&replicas, "replicas", 1, "副本数量")
rootCmd.PersistentFlags().Float64Var(&ratio, "ratio", 0.5, "比例")

// cmd/root.go（定义全局 flag 的部分）
func init() {
	// 默认 kubeconfig = $HOME/.kube/config
	homeDir, _ := os.UserHomeDir()
	defaultKubeconfig := filepath.Join(homeDir, ".kube", "config")

	rootCmd.PersistentFlags().StringVarP(&kubeconfig, "kubeconfig", "k",
		defaultKubeconfig, "kubeconfig 文件路径")
	rootCmd.PersistentFlags().StringVarP(&namespace, "namespace", "n",
		"default", "目标命名空间")
}

// cmd/world.go 的 Run 里
Run: func(cmd *cobra.Command, args []string) {
	fmt.Println("kubeconfig =", kubeconfig)
	fmt.Println("namespace =", namespace)
	// ... 未来在这里调用 client-go 去连集群
},

// cmd/world.go
var source string

func init() {
	helloCmd.AddCommand(worldCmd)
	// 本地 flag：直接挂在 worldCmd.Flags() 上
	worldCmd.Flags().StringVar(&source, "source", "", "数据来源（仅 hello world 可用）")
}

var worldCmd = &cobra.Command{
	Use:        "world",
	Short:      "World 子命令示例",
	Deprecated: "请使用 ask chatgpt 等其他命令替代",
	Run: func(cmd *cobra.Command, args []string) {
		// ... 这个命令仍然能运行，只是提示用户以后可能不再维护
	},
}

k8sco ask chatgpt
# 进入交互终端后你敲：
#   帮我部署一个 Deployment，镜像 image=nginx
#   查询 default 命名空间下的 Pod / Service / Deployment
#   删除 default 命名空间下的 Pod xxx          （作业自己实现）

cobra-cli add ask
cobra-cli add chatgpt --parent askCmd

// cmd/chatgpt.go
package cmd

import (
	"bufio"
	"fmt"
	"os"
	"github.com/spf13/cobra"
)

// StartChat 进入交互对话
func StartChat() error {
	scanner := bufio.NewScanner(os.Stdin) // 从标准输入读取
	fmt.Println("我是 k8sco AI 助手，请问有什么可以帮您?")

	for {
		fmt.Print("> ")                 // 打印输入标识
		if !scanner.Scan() {
			break
		}
		input := scanner.Text()          // 拿到用户输入
		if input == "exit" {
			fmt.Println("再见!")
			break
		}
		if input == "" {
			continue
		}
		fmt.Println("你输入了:", input)  // 先打出来验证能收到输入
	}
	return nil
}

var chatgptCmd = &cobra.Command{
	Use:   "chatgpt",
	Short: "用对话方式操作 K8s 资源",
	Run: func(cmd *cobra.Command, args []string) {
		if err := StartChat(); err != nil {
			fmt.Println("chatgpt err:", err)
		}
	},
}

package openai

import (
	"context"
	"fmt"
	"os"

	"github.com/sashabaranov/go-openai"
)

// Client 是对 OpenAI 兼容客户端的二次封装
type Client struct {
	client  *openai.Client
	context context.Context
}

// NewClient 初始化 OpenAI client
func NewClient() (*Client, error) {
	// 从环境变量读 API Key
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" {
		return nil, fmt.Errorf("OPENAI_API_KEY is not set")
	}

	// 国内网络常需要自定义 baseURL（例如代理 / 某云厂商兼容端点）
	config := openai.DefaultConfig(apiKey)
	config.BaseURL = "https://api.xxx.com/v1"   // 按你的代理/供应商替换

	client := openai.NewClientWithConfig(config)
	return &Client{
		client:  client,
		context: context.Background(),
	}, nil
}

// SendMessage 向大模型发一条消息，返回回复文本
func (c *Client) SendMessage(systemPrompt, userInput string) (string, error) {
	resp, err := c.client.CreateChatCompletion(c.context, openai.ChatCompletionRequest{
		Model: "gpt-4o",
		Messages: []openai.ChatCompletionMessage{
			{Role: "system", Content: systemPrompt},
			{Role: "user",   Content: userInput},
		},
	})
	if err != nil {
		return "", err
	}
	if len(resp.Choices) == 0 {
		return "", fmt.Errorf("no choices returned")
	}
	return resp.Choices[0].Message.Content, nil
}

func processInput(input string) (string, error) {
	client, err := openai.NewClient()
	if err != nil {
		return "", err
	}
	systemPrompt := "你现在是一个 K8s 助手，帮助用户完成任务。"
	return client.SendMessage(systemPrompt, input)
}

resp, err := processInput(input)
if err != nil { /* 处理错误 */ }
fmt.Println(resp)

// 函数 1：生成 K8s YAML 并部署资源
f1 := openai.FunctionDefinition{
	Name:        "generateAndDeployResource",
	Description: "根据用户要求生成 K8s YAML 并部署资源到集群",
	Parameters: map[string]interface{}{
		"type": "object",
		"properties": map[string]interface{}{
			"user_input": {
				"type":        "string",
				"description": "用户输入的原样内容，要求包含资源类型和镜像",
			},
		},
		"required": []string{"user_input"},
	},
}

f2 := openai.FunctionDefinition{
	Name:        "queryResource",
	Description: "查询 K8s 资源",
	Parameters: map[string]interface{}{
		"type": "object",
		"properties": map[string]interface{}{
			"namespace":     {"type": "string", "description": "资源所在的命名空间"},
			"resource_type": {
				"type":        "string",
				"description": "K8s 标准资源类型，例如 pod、deployment、service，请使用标准资源名",
			},
		},
		"required": []string{"namespace", "resource_type"},
	},
}

f3 := openai.FunctionDefinition{
	Name:        "deleteResource",
	Description: "删除 K8s 资源",
	Parameters: map[string]interface{}{
		"type": "object",
		"properties": map[string]interface{}{
			"namespace":     {"type": "string", "description": "资源所在的命名空间"},
			"resource_type": {"type": "string", "description": "K8s 标准资源类型"},
			"resource_name": {"type": "string", "description": "要删除的资源名称"},
		},
		"required": []string{"namespace", "resource_type", "resource_name"},
	},
}

tools := []openai.Tool{
	{Type: "function", Function: &f1},
	{Type: "function", Function: &f2},
	{Type: "function", Function: &f3},
}

resp, err := client.CreateChatCompletion(ctx, openai.ChatCompletionRequest{
	Model:    "gpt-4o",
	Messages: []openai.ChatCompletionMessage{{Role: "user", Content: input}},
	Tools:    tools,
})

msg := resp.Choices[0].Message
if len(msg.ToolCalls) == 0 {
	return "大模型未选择任何工具", nil
}
for _, tc := range msg.ToolCalls {
	fmt.Println("希望调用函数:", tc.Function.Name)
	fmt.Println("参数:", tc.Function.Arguments)
}

func callFunction(name, args string) (string, error) {
	switch name {
	case "generateAndDeployResource":
		return generateAndDeployResource(client, args)   // 真正实现见下一步
	case "queryResource":
		return queryResource(args)                        // 真正实现见下一步
	case "deleteResource":
		return deleteResource(args)                       // 留给你的作业
	default:
		return "", fmt.Errorf("unknown function: %s", name)
	}
}

type generateArgs struct {
	UserInput string `json:"user_input"`
}
var params generateArgs
json.Unmarshal([]byte(args), &params)
// 然后用 params.UserInput 去生成并部署

package k8s

import (
	"path/filepath"
	"strings"

	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
)

type GoK8s struct {
	Clientset      *kubernetes.Clientset
	DynamicClient  dynamic.Interface
	DiscoveryClient discovery.DiscoveryInterface
}

// NewClient 根据 kubeconfig 路径初始化三套客户端
func NewClient(kubeconfig string) (*GoK8s, error) {
	// 处理可能带 ~ 的路径
	if strings.HasPrefix(kubeconfig, "~/") {
		home, _ := os.UserHomeDir()
		kubeconfig = filepath.Join(home, kubeconfig[2:])
	}

	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		return nil, err
	}

	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, err
	}
	dynamicClient, err := dynamic.NewForConfig(config)
	if err != nil {
		return nil, err
	}
	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
	if err != nil {
		return nil, err
	}

	return &GoK8s{Clientset: clientset, DynamicClient: dynamicClient, DiscoveryClient: discoveryClient}, nil
}

func generateAndDeployResource(client *k8s.GoK8s, userInput string) (string, error) {
	// 1. 先用大模型把用户输入"翻译"成一段纯 YAML
	systemPrompt := "你现在是一个 K8s 资源生成器，请根据用户输入生成 K8s YAML，仅输出 YAML，不要包在代码块里。"
	yamlStr, err := openaiClient.SendMessage(systemPrompt, userInput)
	if err != nil {
		return "", err
	}

	// 2. 拿到全集群的 API 资源（GVK -> GVR 的映射需要它）
	_, apiResourceLists, err := client.DiscoveryClient.ServerGroupsAndResources()
	if err != nil {
		return "", err
	}
	restMapper := restmapper.NewDiscoveryRESTMapper(apiResourceLists)

	// 3. 把 YAML 转成 unstructured 对象（因为类型未知，用 unstructured 最通用）
	dec := yaml.NewYAMLOrJSONDecoder(bytes.NewReader([]byte(yamlStr)), 4096)
	var obj map[string]interface{}
	if err := dec.Decode(&obj); err != nil {
		return "", err
	}
	unstructObj := unstructured.Unstructured{Object: obj}

	// 4. 从 GVK 得到 GVR，拿到对应的 REST 接口
	gvk := unstructObj.GroupVersionKind()
	mapping, err := restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		return "", err
	}

	// 5. 确定命名空间（空就用 default）
	namespace := unstructObj.GetNamespace()
	if namespace == "" {
		namespace = "default"
	}

	// 6. 用 dynamic client 创建资源
	created, err := client.DynamicClient.Resource(mapping.Resource).
		Namespace(namespace).
		Create(context.TODO(), &unstructObj, metav1.CreateOptions{})
	if err != nil {
		return "", err
	}
	return "资源创建成功: " + created.GetName(), nil
}

func queryResource(namespace, resourceType string) (string, error) {
	gk8s, err := k8s.NewClient(globalKubeconfig)   // global 里的 kubeconfig
	if err != nil {
		return "", err
	}
	resourceType = strings.ToLower(resourceType)

	var gvr schema.GroupVersionResource
	switch resourceType {
	case "deployment":
		gvr = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
	case "service":
		gvr = schema.GroupVersionResource{Version: "v1", Resource: "services"}
	case "pod":
		gvr = schema.GroupVersionResource{Version: "v1", Resource: "pods"}
	// ... 你可以把几乎所有资源类型都列出来
	default:
		return "", fmt.Errorf("不支持的资源类型: %s", resourceType)
	}

	list, err := gk8s.DynamicClient.Resource(gvr).
		Namespace(namespace).
		List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		return "", err
	}

	var names []string
	for _, item := range list.Items {
		names = append(names, fmt.Sprintf("%s/%s", namespace, item.GetName()))
	}
	return strings.Join(names, "\n"), nil
}

$ ./k8sco ask chatgpt
我是 k8sco AI 助手，请问有什么可以帮您?
> 帮我部署一个 Deployment，镜像是 nginx
资源创建成功: nginx        # dynamic client 已把 YAML 真正 apply 到集群

> 查询 default 命名空间的 pod
default/nginx-7cbb5f7d7f-bz2k8

> 查询 kube-system 的 service
kube-system/kube-dns
...
> exit
再见!

./k8sco analyze events

cobra-cli add analyze
cobra-cli add events --parent analyzeCmd

// getHoldingEventsAndLogs 获取所有 Warning/Error 事件，并按 Pod 分组，同时取日志
func getHoldingEventsAndLogs() (map[string][]string, error) {
	gk8s, err := k8s.NewClient(globalKubeconfig)
	if err != nil {
		return nil, err
	}
	result := make(map[string][]string)

	// 1. 列出 Warning 级别的事件（通过 FieldSelector type=Warning）
	events, err := gk8s.Clientset.CoreV1().Events("").
		List(context.TODO(), metav1.ListOptions{
			FieldSelector: "type=Warning",
		})
	if err != nil {
		return nil, err
	}

	// 2. 遍历事件，按 Pod 分组
	for _, ev := range events.Items {
		podName := ev.InvolvedObject.Name
		namespace := ev.InvolvedObject.Namespace
		message := ev.Message

		// 只处理"对象是 Pod"的事件
		if ev.InvolvedObject.Kind != "Pod" {
			continue
		}

		// 3. 读取该 Pod 的日志
		req := gk8s.Clientset.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{})
		podLogs, err := req.Stream(context.TODO())
		if err != nil {
			continue          // Pod 可能没有日志，跳过
		}
		defer podLogs.Close()

		buf := new(bytes.Buffer)
		if _, err := buf.ReadFrom(podLogs); err != nil {
			continue
		}
		logs := buf.String()
		if logs == "" {
			continue        // 空日志跳过
		}

		// 4. 每组里先放事件信息，再接日志内容
		result[podName] = append(result[podName],
			fmt.Sprintf("event: %s", message),
			fmt.Sprintf("logs:\n%s", logs),
		)
	}
	return result, nil
}

Run: func(cmd *cobra.Command, args []string) {
	eventsLogs, err := getHoldingEventsAndLogs()
	if err != nil {
		fmt.Println("err:", err)
		return
	}
	for pod, infos := range eventsLogs {
		fmt.Println("===== Pod:", pod)
		for _, s := range infos {
			fmt.Println(s)
		}
	}
},

func sendToChatGPT(podInfo map[string][]string) (string, error) {
	client, err := openai.NewClient()
	if err != nil {
		return "", err
	}

	// 拼接：把所有 Pod 的事件和日志合并成一段可读文本
	var combined strings.Builder
	combined.WriteString("找到以下 Pod 的事件和日志:\n")
	for podName, infos := range podInfo {
		combined.WriteString(fmt.Sprintf("\n===== Pod: %s =====\n", podName))
		for _, s := range infos {
			combined.WriteString(s + "\n")
		}
	}
	ctxInfo := combined.String()
	fmt.Println(ctxInfo) // 先打印出来，确认拼对了

	// 构造 Prompt
	messages := []openai.ChatCompletionMessage{
		{Role: "system", Content: "你现在是一位 Kubernetes 专家，请帮助用户诊断 K8s 故障。"},
		{Role: "user", Content: fmt.Sprintf(
			"以下是多个 Pod 的 Warning/Error 事件和对应的日志：\n%s\n"+
			"请主要针对 Pod 的日志给出实质性、可操作的修复建议。", ctxInfo)},
	}

	resp, err := client.CreateChatCompletion(context.TODO(), openai.ChatCompletionRequest{
		Model:    "gpt-4o",
		Messages: messages,
	})
	if err != nil {
		return "", err
	}
	if len(resp.Choices) == 0 {
		return "", fmt.Errorf("no choices")
	}
	return resp.Choices[0].Message.Content, nil
}

./k8sco analyze events