Operator AIOps实战

mkdir demo1 && cd demo1
go mod init spot-pool
kubebuilder init --domain example.com

# 创建一个 API：group = tcluster / iac，version = v1，kind = SpotPool
kubebuilder create api --group iac --version v1 --kind SpotPool \
    --resource --controller

type SpotPoolSpec struct {
	// 初始化腾讯云 SDK 的密钥
	SecretID  string `json:"secretId"`
	SecretKey string `json:"secretKey"`

	Region        string `json:"region"`        // 地域，如 ap-singapore
	AvailableZone string `json:"availableZone"` // 可用区，要跟 VPC/子网匹配
	InstanceType  string `json:"instanceType"`  // 实例类型，如 GN7 系带 T4 GPU

	// 网络相关
	SubnetId        string `json:"subnetId"`
	VPCId           string `json:"vpcId"`
	SecurityGroupId string `json:"securityGroupId"` // 注意复数 groups，敲错会报错
	ImageId         string `json:"imageId"`         // 使用腾讯云官方公开镜像 ID

	InstanceChargeType string `json:"instanceChargeType"` // 计费类型，这里用竞价实例 spot

	// 我们自己关心的控制参数
	MinInstances int32 `json:"minInstances"` // 最小实例数
	MaxInstances int32 `json:"maxInstances"` // 最大实例数
}

type SpotPoolStatus struct {
	Size       int32          `json:"size"`       // 当前实例数
	Conditions []metav1.Condition `json:"conditions"`
	Instances  []InstanceInfo `json:"instances"` // 实例列表
}

type InstanceInfo struct {
	InstanceId string `json:"instanceId"`
	PublicIp   string `json:"publicIp"`
}

func createCVMClient(sp v1.SpotPool) (*cvm.Client, error) {
	credential := common.NewCredential(sp.Spec.SecretID, sp.Spec.SecretKey)
	client, err := cvm.NewClient(credential, sp.Spec.Region,
		&sdkprofile.ClientProfile{SignMethod: "HmacSHA1"})
	if err != nil {
		return nil, err
	}
	return client, nil
}

func getRunningInstanceIds(sp v1.SpotPool) ([]string, error) {
	client, err := createCVMClient(sp)
	if err != nil {
		return nil, err
	}

	request := cvm.NewDescribeInstancesRequest()
	// 镜像、可用区等过滤条件从这里带进去
	response, err := client.DescribeInstances(request)
	if err != nil {
		return nil, err
	}

	var running []string
	for _, instance := range response.Response.InstanceSet {
		state := *instance.InstanceState
		// 只看运行中 / 开通中 / 启动中的实例，跳过其它状态
		if state != "RUNNING" && state != "PENDING" && state != "STARTING" {
			continue
		}
		// 公网 IP 还没分配就返回错误，交给上层稍后重试
		if *instance.PublicIpAddresses[0] == "" {
			return nil, fmt.Errorf("public ip not ready, retry later")
		}
		running = append(running, *instance.InstanceId)
	}
	return running, nil
}

func (r *SpotPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var sp iacv1.SpotPool
	if err := r.Get(ctx, req.NamespacedName, &sp); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	runningIDs, err := getRunningInstanceIds(sp)
	if err != nil {
		// 查询失败（比如公网 IP 还没分配），10 秒后重试
		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
	}
	runningCount := len(runningIDs)

	switch {
	case runningCount < int(sp.Spec.MinInstances):
		delta := int(sp.Spec.MinInstances) - runningCount
		log.FromContext(ctx).Info("need to create instances", "count", delta)
		if err := runInstance(sp, int32(delta)); err != nil {
			return ctrl.Result{RequeueAfter: 40 * time.Second}, nil
		}
	case runningCount > int(sp.Spec.MaxInstances):
		delta := runningCount - int(sp.Spec.MaxInstances)
		log.FromContext(ctx).Info("too many instances, terminate", "count", delta)
		if err := terminateInstance(sp, int32(delta)); err != nil {
			return ctrl.Result{RequeueAfter: 40 * time.Second}, nil
		}
	}

	// 无论如何都要把最新实例状态写回 status，供观测
	_ = r.Status().Update(ctx, &sp)
	// 40 秒后再来一轮，这就是"周期巡检"的来源
	return ctrl.Result{RequeueAfter: 40 * time.Second}, nil
}

func runInstance(sp iacv1.SpotPool, count int32) error {
	client, err := createCVMClient(sp)
	if err != nil {
		return err
	}

	request := cvm.NewRunInstancesRequest()
	request.ImageId = &sp.Spec.ImageId
	request.Placement = &cvm.Placement{Zone: &sp.Spec.AvailableZone}
	request.InstanceChargeType = &sp.Spec.InstanceChargeType
	request.InstanceCount = &count
	// 实例名带时间戳，方便识别是新补出来的
	name := fmt.Sprintf("spot-%d", time.Now().Unix())
	request.InstanceName = &name
	request.InstanceType = &sp.Spec.InstanceType

	// 公网：按流量计费，带宽 100M，需要公网 IP
	request.InternetAccessible = &cvm.InternetAccessible{
		InternetChargeType:      strPtr("TRAFFIC_POSTPAID_BY_HOUR"),
		InternetMaxBandwidthOut: intPtr(100),
		PublicIpAssigned:        boolPtr(true),
	}
	// 登录与安全组
	request.LoginSettings = &cvm.LoginSettings{Password: strPtr("password123")}
	request.SecurityGroupIds = []*string{&sp.Spec.SecurityGroupId}
	// 系统盘 100G 高性能 SSD
	request.SystemDisk = &cvm.SystemDisk{
		DiskType: strPtr("CLOUD_BSSD"), DiskSize: intPtr(100),
	}
	// 网络归属：VPC + 子网
	request.VirtualPrivateCloud = &cvm.VirtualPrivateCloud{
		VpcId: &sp.Spec.VPCId, SubnetId: &sp.Spec.SubnetId,
	}

	response, err := client.RunInstances(request)
	if err != nil {
		return err
	}
	for _, id := range response.Response.InstanceIdSet {
		fmt.Println("created instance:", *id)
	}
	return nil
}

func terminateInstance(sp iacv1.SpotPool, count int32) error {
	client, err := createCVMClient(sp)
	if err != nil {
		return err
	}
	runningIDs, err := getRunningInstanceIds(sp)
	if err != nil {
		return err
	}
	// 只终止前 count 台
	request := cvm.NewTerminateInstancesRequest()
	request.InstanceIds = runningIDs[:count]
	if _, err := client.TerminateInstances(request); err != nil {
		return err
	}
	return nil
}

#!/bin/bash
# 1. 用 Ollama 官方脚本安装
curl -fsSL https://ollama.com/install.sh | sh

# 2. 配置 systemd 实现自启动
cat > /etc/systemd/system/ollama.service <<'EOF'
[Unit]
Description=Ollama
After=network-online.target
[Service]
ExecStart=/usr/local/bin/ollama serve
Restart=always
[Install]
WantedBy=multi-user.target
EOF
systemctl daemon-reload && systemctl enable ollama && systemctl start ollama

# 3. 预加载模型（这一步会联网下载，比较耗时）
ollama pull qwen2:0.5b
ollama create qwen2:0.5b  # 或直接使用已拉取模型

# 4. 把当前系统做成自定义镜像，输出 imageId
echo "完成：请把当前实例导出为自定义镜像，拿到 imageId"

KongGatewayIp string `json:"kongGatewayIp"` // Kong 网关的公网 IP

1. 构造 Kong Admin API URL：http://<kongGatewayIp>:8001/upstreams/ollama-upstream/targets
2. GET 当前所有 target，解析出 {targetId, host(=实例IP:8080)}
3. 遍历实例列表，构造目标 IP 集
4. 对"在 target 里但不在实例里"的 target → DELETE 掉（实例被释放了）
5. 对"在实例里但不在 target 里"的实例 → POST 新增 target（新补的实例加进负载均衡表）

func syncKongUpstream(sp iacv1.SpotPool) error {
	apiBase := "http://" + sp.Spec.KongGatewayIp + ":8001/upstreams/ollama-upstream/targets"

	// 1. 取当前 targets
	resp, err := http.Get(apiBase + "/list")
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("list targets: %d", resp.StatusCode)
	}
	body, _ := io.ReadAll(resp.Body)
	var targetResp struct {
		Data []struct {
			ID   string `json:"id"`
			Host string `json:"host"`
		} `json:"data"`
	}
	if err := json.Unmarshal(body, &targetResp); err != nil {
		return err
	}

	// 2. 从实例状态构建目标 IP 集（Ollama 监听在 8080）
	instanceSet := map[string]bool{}
	instanceList := map[string]string{} // instance publicIp -> target host
	for _, inst := range sp.Status.Instances {
		host := fmt.Sprintf("%s:8080", inst.PublicIp)
		instanceSet[host] = true
		instanceList[inst.PublicIp] = host
	}

	// 3. 剔除：target 存在但实例已释放
	for _, t := range targetResp.Data {
		if !instanceSet[t.Host] {
			delURL := fmt.Sprintf("%s/%s", apiBase, t.ID)
			req, _ := http.NewRequest(http.MethodDelete, delURL, nil)
			client := http.DefaultClient
			if resp, err := client.Do(req); err == nil {
				resp.Body.Close()
			}
		}
	}

	// 4. 新增：实例存在但 target 里没有
	for _, inst := range sp.Status.Instances {
		host := fmt.Sprintf("%s:8080", inst.PublicIp)
		exist := false
		for _, t := range targetResp.Data {
			if t.Host == host {
				exist = true
			}
		}
		if !exist {
			postData, _ := json.Marshal(map[string]string{"target": host})
			http.Post(apiBase, "application/json",
				strings.NewReader(string(postData)))
		}
	}
	return nil
}

curl -X POST http://<kong-ip>:8000/ollama/chat \
  -H "Content-Type: application/json" \
  -d '{"model": "qwen2:0.5b",
       "messages": [{"role": "user", "content": "你好，你是谁"}]}'

Loki 日志系统 ──按时间窗口持续取日志──▶ 命中关键字(error/warning) ──▶ LLM 分析 ──▶ 严重问题 ──▶ 飞书通知
                                                          │
                                            └── 不是所有日志都发给大模型，只按级别过滤 ──┘

type LogPilotSpec struct {
	LokiUrl      string `json:"lokiUrl"`      // Loki 地址
	LokiPromQL   string `json:"lokiPromQL"`   // 查日志的 LogQL 语句，如查 payment 服务
	LLMEndpoint  string `json:"llmEndpoint"`  // LLM 的 OpenAI 兼容接口地址
	LLMToken     string `json:"llmToken"`     // API Key（用自托管 Ollama 可留空）
	LLMModel     string `json:"llmModel"`     // 模型名，如 gpt-4o
	FeishuWebhook string `json:"feishuWebhook"` // 飞书机器人 webhook
}

type LogPilotStatus struct {
	PreviousTimestamp string `json:"previousTimestamp"` // 上一次取日志的时间
}

now := time.Now()
previous := time.Now().Add(-5 * time.Second) // 默认最近 5 秒
if sp.Status.PreviousTimestamp != "" {
	if ts, err := strconv.ParseInt(sp.Status.PreviousTimestamp, 10, 64); err == nil {
		previous = time.Unix(0, ts*int64(time.Millisecond))
	}
}
start := previous.UnixNano() // Loki 需要纳秒级时间戳
end := now.UnixNano()
if start > end {
	// 时间反了，跳过本轮，10 秒后再试
	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
}

lokiURL := fmt.Sprintf("%s/loki/api/v1/query_range?query=%s&start=%d&end=%d",
	sp.Spec.LokiUrl, url.QueryEscape(sp.Spec.LokiPromQL), start, end)
logs, err := queryLoki(lokiURL)

func queryLoki(queryURL string) (string, error) {
	resp, err := http.Get(queryURL)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("loki status: %d", resp.StatusCode)
	}
	body, _ := io.ReadAll(resp.Body)

	var result struct {
		Data struct {
			Result []interface{} `json:"result"`
		} `json:"data"`
	}
	if err := json.Unmarshal(body, &result); err != nil {
		return "", err
	}
	if len(result.Data.Result) == 0 {
		return "", nil // 没有命中日志，返回空
	}
	return string(body), nil
}

func analyzeLogsWithLLM(endpoint, token, model, logs string) (analysedResult, error) {
	conf := openai.DefaultConfig(token)
	conf.BaseURL = endpoint // 重点：替换成我们的 LLM endpoint（可指向自托管 Ollama）
	client := openai.NewClientWithConfig(conf)

	resp, err := client.CreateChatCompletion(context.Background(),
		openai.ChatCompletionRequest{
			Model: model,
			Messages: []openai.ChatCompletionMessage{
				{
					Role: "user",
					Content: fmt.Sprintf(`你现在是一名运维专家，以下日志是从日志系统里获取到的日志，请分析日志错误等级。
如果遇到严重问题（例如：外部系统请求失败、外部系统故障、致命错误、数据库连接异常），请给出简短的建议。
并且对于你认为严重且需要通知运维人员的，请在内容里返回"飞书"标识。以下为日志：
%s`, logs),
				},
			},
		})
	if err != nil {
		return analysedResult{}, err
	}
	content := resp.Choices[0].Message.Content
	res := analysedResult{
		Analysis: content,
		HasError: strings.Contains(strings.ToLower(content), "飞书"),
	}
	return res, nil
}

func sendFeishuAlert(webhook, analysis string) error {
	message := map[string]interface{}{
		"msg_type": "text",
		"content":  map[string]string{"text": analysis},
	}
	body, _ := json.Marshal(message)
	resp, err := http.Post(webhook, "application/json", bytes.NewReader(body))
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("feishu status: %d", resp.StatusCode)
	}
	return nil
}

sp.Status.PreviousTimestamp = fmt.Sprintf("%d", currentTime)
_ = r.Status().Update(ctx)
return ctrl.Result{RequeueAfter: 10 * time.Second}, nil

# 服务 500 错误
- 网关层 500 错误：先检查上游服务健康，联系小王。
- 服务异常：联系小李检查服务与依赖。

# 数据库连接异常
- 连接池耗尽：检查连接配置，提高 connection limit；联系小王提高内存限制。

# memory 相关
- 内存不足导致 OOM：联系小王提高内存限制。

type RagLogPilotSpec struct {
	WorkloadNamespace string `json:"workloadNamespace"` // 取日志的命名空间
	RagflowEndpoint   string `json:"ragflowEndpoint"`   // RAGFlow API 地址（带 /v1/api 后缀）
	RagflowToken      string `json:"ragflowToken"`      // RAGFlow API Key
}

func getAllPodLogs(ctx context.Context, client client.Client, ns string) (string, error) {
	// 1. 列出该 namespace 下所有 Pod
	var pods corev1.PodList
	if err := client.List(ctx, &pods, client.InNamespace(ns)); err != nil {
		return "", err
	}
	// 2. 对每个 Pod 取容器日志（用 controller-runtime 的 client 直接取）
	var sb strings.Builder
	for _, p := range pods.Items {
		req := client.GetLogs(ns, p.Name, &corev1.PodLogOptions{Container: p.Spec.Containers[0].Name})
		rc, err := req.Stream(ctx)
		if err != nil {
			continue
		}
		data, _ := io.ReadAll(rc)
		rc.Close()
		sb.Write(data)
	}
	// 3. 只保留我们关心的行（含 error 等关键字）
	var lines []string
	for _, line := range strings.Split(sb.String(), "\n") {
		if strings.Contains(strings.ToLower(line), "error") {
			lines = append(lines, line)
		}
	}
	return strings.Join(lines, "\n"), nil
}

func queryRagSystem(endpoint, token, logs string) (string, error) {
	const kbPrompt = `以下是获取到的应用日志，请基于运维专家知识库进行解答：
%s
请基于我们内部的专家知识库给出解决方案。如果知识库中没有相关信息，请明确回答"不知道"。`

	// 1. 创建对话，拿到 conversationId
	convURL := endpoint + "/conversation"
	convBody, _ := json.Marshal(map[string]string{"user": "aidesk"})
	req, _ := http.NewRequest(http.MethodPost, convURL, bytes.NewReader(convBody))
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	var cr struct {
		Data struct{ ID string `json:"id"` } `json:"data"`
	}
	_ = json.Unmarshal(b, &cr)
	conversationID := cr.Data.ID

	// 2. 基于这个对话发消息，取 data.answer
	chatBody, _ := json.Marshal(map[string]interface{}{
		"conversation_id": conversationID,
		"message":         fmt.Sprintf(kbPrompt, logs),
		"stream":          false, // 非流式，等它回答完再取
	})
	req2, _ := http.NewRequest(http.MethodPost, endpoint+"/chat", bytes.NewReader(chatBody))
	req2.Header.Set("Authorization", "Bearer "+token)
	req2.Header.Set("Content-Type", "application/json")
	resp2, err := http.DefaultClient.Do(req2)
	if err != nil {
		return "", err
	}
	defer resp2.Body.Close()
	b2, _ := io.ReadAll(resp2.Body)
	var ar struct {
		Data struct{ Answer string `json:"answer"` } `json:"data"`
	}
	if err := json.Unmarshal(b2, &ar); err != nil {
		return "", err
	}
	return ar.Data.Answer, nil
}