从自动化脚本到云原生Operator的实战之路

cmd/                # 可执行文件入口
  agent/            # 运维代理程序
    main.go
internal/           # 内部包（不对外暴露）
  config/
  collector/
pkg/                # 可复用的公共库
  logger/
  httputil/
api/                # API定义（proto或OpenAPI）
configs/            # 配置文件示例
deployments/        # 部署相关（Dockerfile, k8s yaml）
Makefile
go.mod

BINARY_NAME=ops-agent
VERSION=1.0.0

build:
	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags="-s -w" -o bin/${BINARY_NAME} cmd/agent/main.go

test:
	go test -v -race -coverprofile=coverage.out ./...

lint:
	golangci-lint run

docker:
	docker build -t ${BINARY_NAME}:${VERSION} .

package main

import (
	"fmt"
	"sync"
	"time"
)

type Node struct {
	IP   string
	Port int
}

// 模拟健康检查
func checkNode(node Node) bool {
	time.Sleep(100 * time.Millisecond) // 模拟网络请求
	return true // 假设都成功
}

func main() {
	nodes := make([]Node, 1000)
	for i := 0; i < 1000; i++ {
		nodes[i] = Node{IP: fmt.Sprintf("10.0.0.%d", i), Port: 8080}
	}

	// 并发检查，控制并发数20
	concurrency := 20
	ch := make(chan Node, len(nodes))
	var wg sync.WaitGroup

	// 启动worker
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for node := range ch {
				if checkNode(node) {
					fmt.Printf("%s:%d OK\n", node.IP, node.Port)
				}
			}
		}()
	}

	// 投递任务
	for _, node := range nodes {
		ch <- node
	}
	close(ch)
	wg.Wait()
}

package main

import (
	"github.com/gin-gonic/gin"
	"github.com/golang-jwt/jwt/v5"
	"net/http"
)

func AuthMiddleware() gin.HandlerFunc {
	return func(c *gin.Context) {
		token := c.GetHeader("Authorization")
		if token == "" {
			c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing token"})
			return
		}
		// 验证JWT（省略具体逻辑）
		c.Next()
	}
}

func main() {
	r := gin.Default()
	r.Use(gin.Logger())   // 记录请求日志
	r.Use(gin.Recovery()) // panic恢复

	api := r.Group("/api/v1")
	api.Use(AuthMiddleware())
	{
		api.POST("/nodes", registerNode)
		api.GET("/nodes/:id", getNodeStatus)
		api.POST("/tasks", createTask)
	}

	r.Run(":8080")
}

package main

import (
	"context"
	"fmt"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

func main() {
	config, _ := clientcmd.BuildConfigFromFlags("", "/home/user/.kube/config")
	clientset, _ := kubernetes.NewForConfig(config)

	pods, _ := clientset.CoreV1().Pods("default").List(context.TODO(), metav1.ListOptions{})
	for _, pod := range pods.Items {
		fmt.Printf("Pod: %s, Status: %s\n", pod.Name, pod.Status.Phase)
	}
}

type AlertRule struct {
	ID          uint   `gorm:"primaryKey"`
	Name        string `gorm:"uniqueIndex"`
	Metric      string
	Threshold   float64
	Duration    int // 持续分钟数
	Action      string
}

// 创建规则
func CreateRule(db *gorm.DB, rule *AlertRule) error {
	return db.Create(rule).Error
}

// 根据指标查询规则
func GetRulesByMetric(db *gorm.DB, metric string) ([]AlertRule, error) {
	var rules []AlertRule
	err := db.Where("metric = ?", metric).Find(&rules).Error
	return rules, err
}

import "github.com/go-redis/redis/v8"

func acquireLock(client *redis.Client, key string, ttl time.Duration) bool {
	ctx := context.Background()
	// SETNX + EXPIRE
	ok, err := client.SetNX(ctx, key, "locked", ttl).Result()
	if err != nil {
		return false
	}
	return ok
}

func releaseLock(client *redis.Client, key string) {
	client.Del(context.Background(), key)
}

package collector

import (
	"github.com/shirou/gopsutil/v3/cpu"
	"github.com/shirou/gopsutil/v3/mem"
	"time"
)

type Metric struct {
	Timestamp int64
	CPUPercent float64
	MemPercent float64
	DiskUsage  float64
}

func Collect() Metric {
	cpuPerc, _ := cpu.Percent(1*time.Second, false)
	memStat, _ := mem.VirtualMemory()
	return Metric{
		Timestamp:  time.Now().Unix(),
		CPUPercent: cpuPerc[0],
		MemPercent: memStat.UsedPercent,
		// Disk omitted for brevity
	}
}

apiVersion: ops.example.com/v1alpha1
kind: MysqlCluster
metadata:
  name: my-cluster
spec:
  replicas: 1
  version: "8.0"
  storageSize: "100Gi"
  backupSchedule: "0 2 * * *"

func (r *MysqlClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	// 获取CR实例
	var cluster opsv1alpha1.MysqlCluster
	if err := r.Get(ctx, req.NamespacedName, &cluster); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// 确保StatefulSet存在
	sts := &appsv1.StatefulSet{}
	if err := r.Get(ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, sts); err != nil {
		// 不存在则创建
		return r.createStatefulSet(ctx, &cluster)
	}

	// 检查副本数是否一致
	if *sts.Spec.Replicas != cluster.Spec.Replicas {
		// 更新StatefulSet
		patch := client.MergeFrom(sts.DeepCopy())
		sts.Spec.Replicas = &cluster.Spec.Replicas
		if err := r.Patch(ctx, sts, patch); err != nil {
			return ctrl.Result{}, err
		}
	}

	// 检查备份CronJob（略）
	return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
}