概述

运维团队三个人，管着 200 台服务器。每天上班第一件事：挨个 SSH 登录，检查磁盘、内存、CPU、连接数、证书过期时间……一上午就过去了。遇到突发故障，根本来不及巡检，问题已经在用户投诉里炸了。

这不是个别现象。很多中小团队的运维巡检还停留在"人肉+脚本"的阶段——有几个 Shell 脚本，但散落在各个机器上，没人维护，没人知道上次跑是什么时候，输出也没人看。

自动化巡检平台解决的就是这个问题。不是简单的"把脚本集中起来跑"，而是一套完整的体系：插件化的检查项、并发的执行引擎、灵活的告警策略、可读的巡检报告。这篇文章从架构设计到代码实现，讲清楚怎么用 Go 搭一个生产可用的巡检平台。

为什么选 Go 而不是 Python？三个原因：单二进制部署、协程并发模型天然适合巡检场景、交叉编译方便分发到不同架构的服务器。Python 也能做，但在 200 台机器上分发 Python 环境和依赖的痛苦，经历过的人都懂。

巡检平台要解决什么问题

传统巡检的痛点

先看看"人肉巡检"到底有多痛：

痛点具体表现影响
覆盖不全200 台机器只检查了 50 台常用的，剩下的"反正没出过事"隐患积累，暴雷时措手不及
标准不统一A 用 df -h，B 用 df -hT，C 写了个 Python 脚本但只有自己看得懂换人就抓瞎，结果无法横向对比
无历史记录巡检结果写在个人笔记里，离职后带走了无法追溯趋势变化，“上周还好的怎么突然满了”
告警滞后巡检发现磁盘 95%，但没人及时看到从发现到处置间隔几小时甚至几天
人力浪费3 个人每天花 2-3 小时巡检，月度耗时约 200 工时相当于 1.25 个全职人力纯粹在做人肉检查

平台化巡检的核心能力

一个合格的巡检平台需要具备以下能力：

┌─────────────────────────────────────────────────────┐
│                   巡检平台核心能力                     │
├──────────┬──────────┬──────────┬──────────┬─────────┤
│  检查引擎  │  调度系统  │  告警联动  │  报告生成  │  资产管理 │
│          │          │          │          │         │
│ 插件化   │ 定时触发   │ 多级阈值  │ HTML报告  │ 主机清单 │
│ 并发执行  │ 手动触发   │ 告警去重  │ 趋势图表  │ 分组管理 │
│ 超时控制  │ 补偿执行   │ 通知渠道  │ 差异对比  │ 标签体系 │
└──────────┴──────────┴──────────┴──────────┴─────────┘

简单说：配置好检查项和阈值，平台自动在指定时间执行，发现问题立即告警，每次结果都存档可追溯。

架构设计

整体架构

┌──────────────────────────────────────────────────────┐
│                    API / Web UI                       │
│            (配置管理、报告查看、手动触发)               │
├──────────────────────────────────────────────────────┤
│                                                       │
│  ┌─────────────┐  ┌──────────────┐  ┌─────────────┐ │
│  │  调度引擎    │  │  执行引擎     │  │  告警引擎   │ │
│  │             │  │              │  │             │ │
│  │ Cron 触发   │──││ Worker Pool │──││ 阈值判断   │ │
│  │ 手动触发    │  │  并发执行    │  │  去重合并   │ │
│  │ 补偿执行    │  │  超时控制    │  │  多渠道通知  │ │
│  └─────────────┘  └──────────────┘  └─────────────┘ │
│         │                │                │          │
│         │         ┌──────┴──────┐        │          │
│         │         │  插件注册表  │        │          │
│         │         │             │        │          │
│         │         │ disk_check  │        │          │
│         │         │ cpu_check   │        │          │
│         │         │ cert_check  │        │          │
│         │         │ conn_check  │        │          │
│         │         │ ...         │        │          │
│         │         └─────────────┘        │          │
│         │                │                │          │
│  ┌──────┴────────────────┴────────────────┴──────┐  │
│  │                  存储层                         │  │
│  │  SQLite/PostgreSQL (巡检结果、配置、资产)       │  │
│  └───────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────┘

分层职责

层职责技术选型
接入层配置管理、报告展示、手动触发HTTP API + 简单 Web UI
调度层定时触发巡检任务、任务队列管理Go cron 库 + channel
执行层并发执行检查插件、超时控制goroutine + Worker Pool
插件层具体检查项逻辑（磁盘、CPU、证书等）Go interface 插件化
告警层阈值判断、告警去重、通知发送webhook + email + IM
存储层巡检结果、配置、资产信息持久化SQLite（单机）或 PostgreSQL（集群）

为什么用插件化

巡检项是不断变化的。今天检查磁盘，明天要加证书过期检查，后天又要加数据库连接数检查。如果检查逻辑硬编码在主程序里，每次加检查项都要改主程序、重新编译、重新部署。

插件化的好处：新增检查项只需写一个插件文件，注册到插件管理器，重启平台即可生效。不影响现有逻辑，不需要改主程序代码。

Go 的插件化有几种实现方式：

方案优点缺点适用场景
Go plugin 包原生支持，性能好仅 Linux/macOS，编译版本必须一致纯 Go 环境，平台可控
interface + 注册简单可靠，跨平台需要重新编译主程序中小规模，检查项不频繁变更
gRPC 插件语言无关，进程隔离复杂度高，有 RPC 开销多语言插件、强隔离需求
执行外部脚本最灵活，任何语言都能写性能差，依赖外部环境快速原型、遗留脚本复用

综合考虑跨平台和部署便利性，我选择 interface + 注册 方案。虽然新增插件需要重新编译，但 Go 编译速度快，且可以通过配置文件控制哪些插件启用，灵活性够用。

核心代码实现

项目结构

sre-inspector/
├── cmd/
│   └── inspector/
│       └── main.go           # 程序入口
├── internal/
│   ├── config/
│   │   └── config.go         # 配置加载
│   ├── scheduler/
│   │   └── scheduler.go      # 调度引擎
│   ├── executor/
│   │   └── executor.go       # 执行引擎
│   ├── checker/
│   │   ├── checker.go        # Checker 接口定义
│   │   ├── registry.go       # 插件注册表
│   │   ├── disk.go           # 磁盘检查插件
│   │   ├── cpu.go            # CPU 检查插件
│   │   ├── memory.go         # 内存检查插件
│   │   ├── certificate.go    # 证书过期检查插件
│   │   ├── connection.go     # 连接数检查插件
│   │   └── process.go        # 关键进程检查插件
│   ├── alert/
│   │   └── alert.go          # 告警引擎
│   ├── report/
│   │   └── report.go         # 报告生成
│   └── storage/
│       └── storage.go        # 数据存储
├── configs/
│   └── inspector.yaml        # 配置文件
├── go.mod
└── go.sum

Checker 接口定义

插件化的核心是定义好接口。每个检查项都实现这个接口：

// internal/checker/checker.go

package checker

import (
    "context"
    "time"
)

// CheckResult 单次检查的结果
type CheckResult struct {
    CheckerName  string            // 检查项名称
    Target       string            // 检查目标（IP、主机名等）
    Status       Status            // 状态：正常/警告/严重
    Message      string            // 结果描述
    Metrics      map[string]float64 // 指标数据（磁盘使用率、CPU负载等）
    CheckedAt    time.Time         // 检查时间
    Duration     time.Duration     // 检查耗时
}

// Status 检查状态
type Status int

const (
    StatusOK      Status = 0 // 正常
    StatusWarning Status = 1 // 警告
    StatusCritical Status = 2 // 严重
    StatusUnknown Status = 3 // 未知（检查失败）
)

func (s Status) String() string {
    switch s {
    case StatusOK:
        return "OK"
    case StatusWarning:
        return "WARNING"
    case StatusCritical:
        return "CRITICAL"
    default:
        return "UNKNOWN"
    }
}

// Checker 检查器接口，所有插件必须实现
type Checker interface {
    // Name 返回检查器名称
    Name() string

    // Description 返回检查器描述
    Description() string

    // Check 执行检查
    // ctx 用于超时控制和取消
    // target 是检查目标（主机IP、主机名等）
    // params 是检查参数（阈值等）
    Check(ctx context.Context, target string, params map[string]string) CheckResult
}

插件注册表

// internal/checker/registry.go

package checker

import (
    "fmt"
    "sync"
)

// Registry 插件注册表
type Registry struct {
    mu       sync.RWMutex
    checkers map[string]Checker
}

// NewRegistry 创建注册表实例
func NewRegistry() *Registry {
    return &Registry{
        checkers: make(map[string]Checker),
    }
}

// Register 注册一个检查器
func (r *Registry) Register(c Checker) error {
    r.mu.Lock()
    defer r.mu.Unlock()

    name := c.Name()
    if _, exists := r.checkers[name]; exists {
        return fmt.Errorf("checker %q already registered", name)
    }
    r.checkers[name] = c
    return nil
}

// Get 获取指定检查器
func (r *Registry) Get(name string) (Checker, bool) {
    r.mu.RLock()
    defer r.mu.RUnlock()
    c, ok := r.checkers[name]
    return c, ok
}

// List 返回所有已注册的检查器名称
func (r *Registry) List() []string {
    r.mu.RLock()
    defer r.mu.RUnlock()
    names := make([]string, 0, len(r.checkers))
    for name := range r.checkers {
        names = append(names, name)
    }
    return names
}

// RegisterBuiltin 注册内置检查器
func (r *Registry) RegisterBuiltin() {
    r.Register(&DiskChecker{})
    r.Register(&CPUChecker{})
    r.Register(&MemoryChecker{})
    r.Register(&CertificateChecker{})
    r.Register(&ConnectionChecker{})
    r.Register(&ProcessChecker{})
}

磁盘检查插件实现

以磁盘检查为例，展示一个完整插件的写法：

// internal/checker/disk.go

package checker

import (
    "context"
    "fmt"
    "os/exec"
    "strconv"
    "strings"
    "time"
)

// DiskChecker 磁盘使用率检查器
type DiskChecker struct{}

func (d *DiskChecker) Name() string        { return "disk" }
func (d *DiskChecker) Description() string  { return "检查磁盘分区使用率" }

func (d *DiskChecker) Check(ctx context.Context, target string, params map[string]string) CheckResult {
    start := time.Now()

    // 解析阈值参数，默认警告80% 严重95%
    warnThreshold := parseFloatParam(params, "warning", 80.0)
    critThreshold := parseFloatParam(params, "critical", 95.0)

    // 通过 SSH 执行远程命令（如果 target 不是本机）
    // 这里用 SSH 简化演示，实际可以用 SSH 连接池
    cmd := exec.CommandContext(ctx, "ssh", "-o", "ConnectTimeout=5", target, "df -h --output=pcent,target -x tmpfs -x devtmpfs")
    output, err := cmd.CombinedOutput()
    if err != nil {
        return CheckResult{
            CheckerName: d.Name(),
            Target:      target,
            Status:      StatusUnknown,
            Message:      fmt.Sprintf("SSH 执行失败: %v, output: %s", err, string(output)),
            CheckedAt:    start,
            Duration:     time.Since(start),
        }
    }

    // 解析 df 输出，找出使用率最高的分区
    var maxUsage float64
    var maxPartitions []string
    lines := strings.Split(strings.TrimSpace(string(output)), "\n")
    for i, line := range lines {
        if i == 0 {
            continue // 跳过表头
        }
        fields := strings.Fields(line)
        if len(fields) < 2 {
            continue
        }
        // fields[0] = "45%"  fields[1] = "/data"
        usageStr := strings.TrimSuffix(fields[0], "%")
        usage, err := strconv.ParseFloat(usageStr, 64)
        if err != nil {
            continue
        }
        if usage > maxUsage {
            maxUsage = usage
            maxPartitions = append(maxPartitions[:0], fields[1])
        } else if usage == maxUsage {
            maxPartitions = append(maxPartitions, fields[1])
        }
    }

    // 根据阈值判断状态
    status := StatusOK
    message := fmt.Sprintf("磁盘使用率最高: %.1f%% (%s)", maxUsage, strings.Join(maxPartitions, ", "))
    if maxUsage >= critThreshold {
        status = StatusCritical
        message = fmt.Sprintf("磁盘使用率严重: %.1f%% (%s) >= %.0f%%", maxUsage, strings.Join(maxPartitions, ", "), critThreshold)
    } else if maxUsage >= warnThreshold {
        status = StatusWarning
        message = fmt.Sprintf("磁盘使用率告警: %.1f%% (%s) >= %.0f%%", maxUsage, strings.Join(maxPartitions, ", "), warnThreshold)
    }

    return CheckResult{
        CheckerName: d.Name(),
        Target:      target,
        Status:      status,
        Message:      message,
        Metrics:      map[string]float64{"disk_usage_percent": maxUsage},
        CheckedAt:    start,
        Duration:     time.Since(start),
    }
}

func parseFloatParam(params map[string]string, key string, defaultVal float64) float64 {
    if val, ok := params[key]; ok {
        if f, err := strconv.ParseFloat(val, 64); err == nil {
            return f
        }
    }
    return defaultVal
}

执行引擎：并发控制

巡检 200 台机器，串行执行要等半小时。并发执行是必须的，但不能无限并发——SSH 连接数太多会把目标机器的 sshd 撑爆，也会把本机 fd 耗尽。

// internal/executor/executor.go

package executor

import (
    "context"
    "log"
    "sync"
    "time"

    "sre-inspector/internal/checker"
)

// Executor 执行引擎
type Executor struct {
    registry    *checker.Registry
    maxWorkers  int           // 最大并发数
    timeout     time.Duration // 单次检查超时
}

// NewExecutor 创建执行引擎
func NewExecutor(registry *checker.Registry, maxWorkers int, timeout time.Duration) *Executor {
    return &Executor{
        registry:   registry,
        maxWorkers: maxWorkers,
        timeout:    timeout,
    }
}

// Task 单个巡检任务
type Task struct {
    CheckerName string            // 检查器名称
    Target      string            // 目标主机
    Params      map[string]string // 检查参数
}

// RunBatch 批量执行巡检任务
func (e *Executor) RunBatch(tasks []Task) []checker.CheckResult {
    // 用带缓冲的 channel 做任务队列
    taskCh := make(chan Task, len(tasks))
    resultCh := make(chan checker.CheckResult, len(tasks))

    // 填充任务队列
    for _, task := range tasks {
        taskCh <- task
    }
    close(taskCh)

    // 启动 worker pool
    var wg sync.WaitGroup
    for i := 0; i < e.maxWorkers; i++ {
        wg.Add(1)
        go e.worker(&wg, taskCh, resultCh)
    }

    // 等待所有 worker 完成
    go func() {
        wg.Wait()
        close(resultCh)
    }()

    // 收集结果
    results := make([]checker.CheckResult, 0, len(tasks))
    for result := range resultCh {
        results = append(results, result)
    }
    return results
}

// worker 工作协程
func (e *Executor) worker(wg *sync.WaitGroup, taskCh <-chan Task, resultCh chan<- checker.CheckResult) {
    defer wg.Done()
    for task := range taskCh {
        result := e.runSingle(task)
        resultCh <- result
    }
}

// runSingle 执行单个检查任务
func (e *Executor) runSingle(task Task) checker.CheckResult {
    c, ok := e.registry.Get(task.CheckerName)
    if !ok {
        return checker.CheckResult{
            CheckerName: task.CheckerName,
            Target:      task.Target,
            Status:      checker.StatusUnknown,
            Message:      fmt.Sprintf("检查器 %q 未注册", task.CheckerName),
        }
    }

    // 带超时的 context
    ctx, cancel := context.WithTimeout(context.Background(), e.timeout)
    defer cancel()

    // 执行检查
    result := c.Check(ctx, task.Target, task.Params)

    // 记录慢检查
    if result.Duration > 5*time.Second {
        log.Printf("[WARN] 检查耗时较长: checker=%s target=%s duration=%s",
            task.CheckerName, task.Target, result.Duration)
    }

    return result
}

调度引擎

// internal/scheduler/scheduler.go

package scheduler

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "sre-inspector/internal/checker"
    "sre-inspector/internal/executor"
)

// ScheduledJob 定时巡检任务
type ScheduledJob struct {
    Name         string                   // 任务名称
    Cron         string                   // cron 表达式
    Checkers     []string                 // 要执行的检查器列表
    Targets      []string                 // 目标主机列表
    Params       map[string]map[string]string // 每个检查器的参数
}

// Scheduler 调度引擎
type Scheduler struct {
    jobs     []*ScheduledJob
    executor *executor.Executor
    alertCh  chan checker.CheckResult // 告警通道
    mu       sync.Mutex
    running  bool
}

// NewScheduler 创建调度器
func NewScheduler(exec *executor.Executor, alertCh chan checker.CheckResult) *Scheduler {
    return &Scheduler{
        executor: exec,
        alertCh:  alertCh,
    }
}

// AddJob 添加定时任务
func (s *Scheduler) AddJob(job *ScheduledJob) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.jobs = append(s.jobs, job)
}

// Start 启动调度器
func (s *Scheduler) Start(ctx context.Context) {
    s.mu.Lock()
    s.running = true
    s.mu.Unlock()

    for _, job := range s.jobs {
        go s.runJob(ctx, job)
    }
    log.Printf("调度器已启动，共 %d 个定时任务", len(s.jobs))
}

// runJob 运行单个定时任务
func (s *Scheduler) runJob(ctx context.Context, job *ScheduledJob) {
    // 解析 cron 表达式，计算下次执行时间
    // 这里用简化版：按固定间隔执行
    interval, err := parseCronInterval(job.Cron)
    if err != nil {
        log.Printf("[ERROR] 解析 cron 表达式失败: job=%s cron=%s err=%v", job.Name, job.Cron, err)
        return
    }

    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    log.Printf("[INFO] 任务 %s 已注册，执行间隔: %v", job.Name, interval)

    for {
        select {
        case <-ctx.Done():
            log.Printf("[INFO] 任务 %s 已停止", job.Name)
            return
        case <-ticker.C:
            log.Printf("[INFO] 开始执行巡检任务: %s", job.Name)
            s.executeJob(job)
        }
    }
}

// executeJob 执行一次完整的巡检
func (s *Scheduler) executeJob(job *ScheduledJob) {
    start := time.Now()

    // 构建任务列表：检查器 × 目标主机 的笛卡尔积
    var tasks []executor.Task
    for _, checkerName := range job.Checkers {
        params := job.Params[checkerName]
        for _, target := range job.Targets {
            tasks = append(tasks, executor.Task{
                CheckerName: checkerName,
                Target:      target,
                Params:      params,
            })
        }
    }

    log.Printf("[INFO] 任务 %s: 共 %d 个检查项", job.Name, len(tasks))

    // 批量执行
    results := s.executor.RunBatch(tasks)

    // 统计
    var okCount, warnCount, critCount, unknownCount int
    for _, r := range results {
        switch r.Status {
        case checker.StatusOK:
            okCount++
        case checker.StatusWarning:
            warnCount++
            s.alertCh <- r // 警告级推送到告警通道
        case checker.StatusCritical:
            critCount++
            s.alertCh <- r // 严重级推送到告警通道
        default:
            unknownCount++
            s.alertCh <- r // 未知也推送，可能是检查失败
        }
    }

    log.Printf("[INFO] 任务 %s 完成: 耗时=%s 正常=%d 警告=%d 严重=%d 未知=%d",
        job.Name, time.Since(start), okCount, warnCount, critCount, unknownCount)
}

// parseCronInterval 简化版 cron 解析
// 支持格式: "*/30 * * * *" → 30分钟间隔
// 这里简化处理，实际建议用 robfig/cron 库
func parseCronInterval(cronExpr string) (time.Duration, error) {
    // 简化实现：只支持 "every Nm" 或 "every Nh" 格式
    var num int
    var unit string
    _, err := fmt.Sscanf(cronExpr, "every %d%c", &num, &unit)
    if err != nil {
        // 默认每小时执行一次
        return time.Hour, nil
    }
    switch unit {
    case 'm':
        return time.Duration(num) * time.Minute, nil
    case 'h':
        return time.Duration(num) * time.Hour, nil
    default:
        return time.Hour, fmt.Errorf("unsupported time unit: %c", unit)
    }
}

告警引擎

// internal/alert/alert.go

package alert

import (
    "fmt"
    "log"
    "strings"
    "sync"
    "time"

    "sre-inspector/internal/checker"
)

// AlertConfig 告警配置
type AlertConfig struct {
    WebhookURL    string   // Webhook 通知地址
    EmailTo       []string // 邮件通知收件人
    DingtalkToken string   // 钉钉机器人 token
    RepeatInterval time.Duration // 重复告警抑制间隔
}

// AlertEngine 告警引擎
type AlertEngine struct {
    config       AlertConfig
    recentAlerts sync.Map // 去重缓存: key="checker:target" → lastAlertTime
}

// NewAlertEngine 创建告警引擎
func NewAlertEngine(config AlertConfig) *AlertEngine {
    return &AlertEngine{config: config}
}

// Process 处理告警结果
func (a *AlertEngine) Process(result checker.CheckResult) {
    if result.Status == checker.StatusOK {
        return // 正常结果不告警
    }

    key := fmt.Sprintf("%s:%s", result.CheckerName, result.Target)

    // 去重：同一检查项+目标在 RepeatInterval 内不重复告警
    if lastTime, ok := a.recentAlerts.Load(key); ok {
        if time.Since(lastTime.(time.Time)) < a.config.RepeatInterval {
            return
        }
    }
    a.recentAlerts.Store(key, time.Now())

    // 构建告警消息
    title, message := a.buildMessage(result)
    log.Printf("[ALERT] %s: %s", title, message)

    // 发送通知（按配置的渠道）
    if a.config.WebhookURL != "" {
        a.sendWebhook(title, message, result)
    }
    if a.config.DingtalkToken != "" {
        a.sendDingtalk(title, message)
    }
}

// buildMessage 构建告警消息
func (a *AlertEngine) buildMessage(result checker.CheckResult) (string, string) {
    severity := "WARNING"
    icon := "⚠️"
    if result.Status == checker.StatusCritical {
        severity = "CRITICAL"
        icon = "🔴"
    } else if result.Status == checker.StatusUnknown {
        severity = "UNKNOWN"
        icon = "❓"
    }

    title := fmt.Sprintf("%s [%s] 巡检告警: %s @ %s", icon, severity, result.CheckerName, result.Target)

    var sb strings.Builder
    sb.WriteString(fmt.Sprintf("**检查项**: %s\n", result.CheckerName))
    sb.WriteString(fmt.Sprintf("**目标主机**: %s\n", result.Target))
    sb.WriteString(fmt.Sprintf("**状态**: %s\n", severity))
    sb.WriteString(fmt.Sprintf("**详情**: %s\n", result.Message))
    sb.WriteString(fmt.Sprintf("**检查时间**: %s\n", result.CheckedAt.Format("2006-01-02 15:04:05")))
    sb.WriteString(fmt.Sprintf("**耗时**: %s\n", result.Duration))

    if len(result.Metrics) > 0 {
        sb.WriteString("**指标**:\n")
        for k, v := range result.Metrics {
            sb.WriteString(fmt.Sprintf("  - %s: %.2f\n", k, v))
        }
    }

    return title, sb.String()
}

// sendWebhook 发送 Webhook 通知
func (a *AlertEngine) sendWebhook(title, message string, result checker.CheckResult) {
    // 实际实现用 http.Post 发送 JSON
    // 省略 HTTP 调用细节
    log.Printf("[WEBHOOK] 发送告警到 %s: %s", a.config.WebhookURL, title)
}

// sendDingtalk 发送钉钉通知
func (a *AlertEngine) sendDingtalk(title, message string) {
    // 实际实现调用钉钉机器人 API
    log.Printf("[DINGTALK] 发送告警: %s", title)
}

// Start 启动告警引擎，监听结果通道
func (a *AlertEngine) Start(alertCh <-chan checker.CheckResult) {
    for result := range alertCh {
        a.Process(result)
    }
}

配置文件

# configs/inspector.yaml

# 执行引擎配置
executor:
  max_workers: 20          # 最大并发数
  timeout: 30s             # 单次检查超时

# 告警配置
alert:
  webhook_url: "https://hooks.example.com/inspector"
  dingtalk_token: ""
  repeat_interval: 30m     # 同一告警30分钟内不重复

# 存储配置
storage:
  type: sqlite             # sqlite 或 postgres
  dsn: "/var/lib/inspector/inspector.db"

# 资产清单
targets:
  - host: "10.0.0.5"
    name: "web-01"
    tags: ["web", "production"]
  - host: "10.0.0.6"
    name: "web-02"
    tags: ["web", "production"]
  - host: "10.0.0.10"
    name: "db-01"
    tags: ["database", "production"]

# 检查器配置
checkers:
  - name: disk
    enabled: true
    params:
      warning: "80"
      critical: "95"
  
  - name: cpu
    enabled: true
    params:
      warning: "70"
      critical: "90"
      interval: "5m"   # 取5分钟平均负载
  
  - name: memory
    enabled: true
    params:
      warning: "80"
      critical: "95"
  
  - name: certificate
    enabled: true
    params:
      warning: "30"    # 30天前告警
      critical: "7"     # 7天前严重
      domains: "api.example.com,admin.example.com"
  
  - name: connection
    enabled: true
    params:
      warning: "5000"
      critical: "10000"
  
  - name: process
    enabled: true
    params:
      processes: "nginx,mysql,redis"  # 必须存活的进程列表

# 定时任务
jobs:
  - name: "每日全面巡检"
    schedule: "every 6h"
    checkers: ["disk", "cpu", "memory", "certificate", "connection", "process"]
    target_tags: ["production"]
  
  - name: "数据库专项巡检"
    schedule: "every 1h"
    checkers: ["disk", "memory", "process"]
    target_tags: ["database"]
  
  - name: "证书每日检查"
    schedule: "every 24h"
    checkers: ["certificate"]
    target_tags: ["production"]

主程序入口

// cmd/inspector/main.go

package main

import (
    "context"
    "flag"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "sre-inspector/internal/alert"
    "sre-inspector/internal/checker"
    "sre-inspector/internal/config"
    "sre-inspector/internal/executor"
    "sre-inspector/internal/report"
    "sre-inspector/internal/scheduler"
    "sre-inspector/internal/storage"
)

func main() {
    configPath := flag.String("config", "configs/inspector.yaml", "配置文件路径")
    once := flag.Bool("once", false, "执行一次巡检后退出（不做定时调度）")
    flag.Parse()

    // 加载配置
    cfg, err := config.Load(*configPath)
    if err != nil {
        log.Fatalf("加载配置失败: %v", err)
    }

    // 初始化插件注册表
    registry := checker.NewRegistry()
    registry.RegisterBuiltin()
    log.Printf("已注册检查器: %v", registry.List())

    // 创建告警通道
    alertCh := make(chan checker.CheckResult, 1000)

    // 初始化执行引擎
    exec := executor.NewExecutor(
        registry,
        cfg.Executor.MaxWorkers,
        cfg.Executor.Timeout,
    )

    // 初始化告警引擎
    alertEngine := alert.NewAlertEngine(cfg.Alert)
    go alertEngine.Start(alertCh)

    // 初始化存储
    store, err := storage.New(cfg.Storage)
    if err != nil {
        log.Fatalf("初始化存储失败: %v", err)
    }
    defer store.Close()

    // 初始化报告生成器
    reportGen := report.NewGenerator(store)

    // 如果是一次性执行模式
    if *once {
        runOnce(cfg, exec, alertCh, store, reportGen)
        return
    }

    // 初始化调度器
    sched := scheduler.NewScheduler(exec, alertCh)
    for _, jobCfg := range cfg.Jobs {
        job := buildJobFromConfig(jobCfg, cfg)
        sched.AddJob(job)
    }

    // 启动调度器
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    sched.Start(ctx)

    log.Println("巡检平台已启动，按 Ctrl+C 退出")

    // 等待退出信号
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    <-sigCh

    log.Println("正在关闭...")
    cancel()
    time.Sleep(2 * time.Second) // 给 goroutine 留时间优雅退出
    log.Println("已退出")
}

// runOnce 执行一次巡检
func runOnce(cfg *config.Config, exec *executor.Executor, alertCh chan checker.CheckResult, store *storage.Storage, reportGen *report.Generator) {
    log.Println("开始一次性巡检...")
    start := time.Now()

    // 构建任务
    var tasks []executor.Task
    for _, checkerCfg := range cfg.Checkers {
        if !checkerCfg.Enabled {
            continue
        }
        for _, target := range cfg.Targets {
            tasks = append(tasks, executor.Task{
                CheckerName: checkerCfg.Name,
                Target:      target.Host,
                Params:      checkerCfg.Params,
            })
        }
    }

    results := exec.RunBatch(tasks)

    // 存储结果
    store.SaveResults(results)

    // 发送告警
    for _, r := range results {
        if r.Status != checker.StatusOK {
            alertCh <- r
        }
    }

    // 生成报告
    reportPath := reportGen.GenerateHTML(results)
    log.Printf("巡检完成: 耗时=%s 报告=%s", time.Since(start), reportPath)

    // 打印摘要
    var ok, warn, crit, unk int
    for _, r := range results {
        switch r.Status {
        case checker.StatusOK:
            ok++
        case checker.StatusWarning:
            warn++
        case checker.StatusCritical:
            crit++
        default:
            unk++
        }
    }
    log.Printf("汇总: 正常=%d 警告=%d 严重=%d 未知=%d", ok, warn, crit, unk)
}

巡检报告生成

巡检的最终产出是报告。报告不是给机器看的，是给人看的——所以要直观、可读、能快速定位问题。

HTML 报告结构

// internal/report/report.go

package report

import (
    "fmt"
    "html/template"
    "os"
    "path/filepath"
    "time"

    "sre-inspector/internal/checker"
)

type Generator struct {
    tmpl *template.Template
}

func NewGenerator(store *storage.Storage) *Generator {
    return &Generator{}
}

// ReportData 报告数据
type ReportData struct {
    GeneratedAt  time.Time
    TotalCount   int
    OKCount      int
    WarningCount int
    CriticalCount int
    UnknownCount int
    Results      []checker.CheckResult
    Summary      map[string]map[string]int // 按主机分组的统计
}

// GenerateHTML 生成 HTML 报告
func (g *Generator) GenerateHTML(results []checker.CheckResult) string {
    data := g.prepareData(results)

    tmpl := template.Must(template.New("report").Parse(reportTemplate))

    filename := fmt.Sprintf("inspection_%s.html", time.Now().Format("20060102_150405"))
    filepath := filepath.Join("/var/lib/inspector/reports", filename)

    os.MkdirAll(filepath, 0755)
    f, err := os.Create(filepath)
    if err != nil {
        return ""
    }
    defer f.Close()

    tmpl.Execute(f, data)
    return filepath
}

func (g *Generator) prepareData(results []checker.CheckResult) ReportData {
    data := ReportData{
        GeneratedAt: time.Now(),
        TotalCount:  len(results),
        Results:     results,
        Summary:     make(map[string]map[string]int),
    }

    for _, r := range results {
        if data.Summary[r.Target] == nil {
            data.Summary[r.Target] = make(map[string]int)
        }
        data.Summary[r.Target][r.Status.String()]++
        switch r.Status {
        case checker.StatusOK:
            data.OKCount++
        case checker.StatusWarning:
            data.WarningCount++
        case checker.StatusCritical:
            data.CriticalCount++
        default:
            data.UnknownCount++
        }
    }
    return data
}

// reportTemplate HTML 报告模板（简化版）
const reportTemplate = `
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>巡检报告 - {{.GeneratedAt.Format "2006-01-02 15:04"}}</title>
    <style>
        body { font-family: -apple-system, sans-serif; margin: 40px; }
        .summary { display: flex; gap: 20px; margin-bottom: 30px; }
        .card { padding: 20px; border-radius: 8px; color: white; }
        .ok { background: #4caf50; }
        .warning { background: #ff9800; }
        .critical { background: #f44336; }
        .unknown { background: #9e9e9e; }
        table { border-collapse: collapse; width: 100%; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background: #f5f5f5; }
        .status-ok { color: #4caf50; }
        .status-warning { color: #ff9800; }
        .status-critical { color: #f44336; }
        .status-unknown { color: #9e9e9e; }
    </style>
</head>
<body>
    <h1>巡检报告</h1>
    <p>生成时间: {{.GeneratedAt.Format "2006-01-02 15:04:05"}}</p>

    <div class="summary">
        <div class="card ok"><h2>{{.OKCount}}</h2><p>正常</p></div>
        <div class="card warning"><h2>{{.WarningCount}}</h2><p>警告</p></div>
        <div class="card critical"><h2>{{.CriticalCount}}</h2><p>严重</p></div>
        <div class="card unknown"><h2>{{.UnknownCount}}</h2><p>未知</p></div>
    </div>

    <h2>详细结果</h2>
    <table>
        <tr>
            <th>检查项</th>
            <th>目标主机</th>
            <th>状态</th>
            <th>详情</th>
            <th>耗时</th>
        </tr>
        {{range .Results}}
        <tr>
            <td>{{.CheckerName}}</td>
            <td>{{.Target}}</td>
            <td class="status-{{.Status.String | toLower}}">{{.Status}}</td>
            <td>{{.Message}}</td>
            <td>{{.Duration}}</td>
        </tr>
        {{end}}
    </table>
</body>
</html>
`

生产部署方案

二进制部署

# 交叉编译
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o bin/inspector ./cmd/inspector

# 部署到服务器
scp bin/inspector configs/inspector.yaml user@server:/opt/inspector/

# 创建 systemd 服务
cat > /etc/systemd/system/inspector.service << 'EOF'
[Unit]
Description=SRE Inspection Platform
After=network.target

[Service]
Type=simple
User=inspector
WorkingDirectory=/opt/inspector
ExecStart=/opt/inspector/inspector -config /opt/inspector/configs/inspector.yaml
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable inspector
systemctl start inspector

SSH 免密配置

巡检需要 SSH 到目标机器执行命令。用密钥认证，别用密码：

# 在巡检服务器上生成专用密钥
ssh-keygen -t ed25519 -f /home/inspector/.ssh/inspector_key -N ""

# 分发公钥到目标机器
for host in 10.0.0.{5..50}; do
    ssh-copy-id -i /home/inspector/.ssh/inspector_key.pub inspector@$host
done

# 配置 SSH 别名，简化连接
cat >> /home/inspector/.ssh/config << 'EOF'
Host *
    IdentityFile ~/.ssh/inspector_key
    StrictHostKeyChecking no
    ConnectTimeout 5
    ServerAliveInterval 30
EOF

安全注意事项

  1. 最小权限原则：巡检用户只给只读权限，别用 root 跑巡检。创建专用的 inspector 用户，sudo 白名单只允许特定只读命令。
# /etc/sudoers.d/inspector
inspector ALL=(ALL) NOPASSWD: /usr/bin/df, /usr/bin/free, /usr/bin/uptime, /usr/bin/ss, /usr/bin/ps

  1. SSH 密钥保护：巡检服务器的私钥泄露 = 所有目标机器沦陷。私钥文件权限 600，最好放在硬件安全模块或密钥管理服务中。

  2. 巡检数据敏感：巡检结果包含主机信息、进程列表、网络连接等，属于敏感运维数据。报告文件存放在受控目录，定期清理过期报告。

  3. 网络隔离：巡检服务器能访问所有目标机器，本身就是高价值攻击目标。放在管理网络中，限制入站访问。

常见检查项实现思路

证书过期检查

// 核心逻辑：用 TLS 连接目标域名，读取证书链，检查 NotAfter 字段
func checkCertificate(domain string, warnDays, critDays int) CheckResult {
    conn, err := tls.Dial("tcp", domain+":443", &tls.Config{
        InsecureSkipVerify: true, // 我们就是要检查过期证书，不能因为过期就连不上
    })
    if err != nil {
        return CheckResult{Status: StatusUnknown, Message: fmt.Sprintf("TLS 连接失败: %v", err)}
    }
    defer conn.Close()

    cert := conn.ConnectionState().PeerCertificates[0]
    daysLeft := int(time.Until(cert.NotAfter).Hours() / 24)

    status := StatusOK
    message := fmt.Sprintf("证书还有 %d 天过期 (到期日: %s)", daysLeft, cert.NotAfter.Format("2006-01-02"))

    if daysLeft <= critDays {
        status = StatusCritical
        message = fmt.Sprintf("证书即将过期！仅剩 %d 天 (到期日: %s)", daysLeft, cert.NotAfter.Format("2006-01-02"))
    } else if daysLeft <= warnDays {
        status = StatusWarning
        message = fmt.Sprintf("证书将在 %d 天后过期 (到期日: %s)", daysLeft, cert.NotAfter.Format("2006-01-02"))
    }

    return CheckResult{
        Status:  status,
        Message:  message,
        Metrics:  map[string]float64{"days_until_expiry": float64(daysLeft)},
    }
}

关键进程存活检查

// 通过 SSH 执行 pgrep 检查进程是否存在
func checkProcess(ctx context.Context, target string, processNames []string) CheckResult {
    var missing []string
    for _, name := range processNames {
        cmd := exec.CommandContext(ctx, "ssh", target, fmt.Sprintf("pgrep -x %s", name))
        if err := cmd.Run(); err != nil {
            missing = append(missing, name)
        }
    }

    if len(missing) > 0 {
        return CheckResult{
            Status:  StatusCritical,
            Message:  fmt.Sprintf("以下进程未运行: %s", strings.Join(missing, ", ")),
        }
    }
    return CheckResult{Status: StatusOK, Message: "所有关键进程正常运行"}
}

TCP 连接数检查

// 通过 SSH 执行 ss 统计当前连接数
func checkConnectionCount(ctx context.Context, target string, warn, crit int) CheckResult {
    cmd := exec.CommandContext(ctx, "ssh", target, "ss -tn state established | wc -l")
    output, err := cmd.Output()
    if err != nil {
        return CheckResult{Status: StatusUnknown, Message: fmt.Sprintf("执行失败: %v", err)}
    }

    count := 0
    fmt.Sscanf(strings.TrimSpace(string(output)), "%d", &count)

    status := StatusOK
    message := fmt.Sprintf("当前 ESTABLISHED 连接数: %d", count)
    if count >= crit {
        status = StatusCritical
        message = fmt.Sprintf("连接数严重: %d >= %d", count, crit)
    } else if count >= warn {
        status = StatusWarning
        message = fmt.Sprintf("连接数告警: %d >= %d", count, warn)
    }

    return CheckResult{
        Status:  status,
        Message:  message,
        Metrics:  map[string]float64{"established_connections": float64(count)},
    }
}

与现有监控系统的关系

巡检平台和 Prometheus 这类监控系统不冲突，定位不同：

维度巡检平台Prometheus
检查频率低频（每小时/每天）高频（每15-60秒）
检查范围综合性（磁盘+证书+进程+配置）指标为主（CPU/内存/流量）
告警方式巡检报告+即时告警实时告警
适用场景定期体检、合规检查、配置审计实时监控、动态告警
数据保留长期（月度/季度趋势）中短期（通常15-90天）

简单说：Prometheus 是心电图，7x24 实时监测；巡检平台是年度体检，定期全面检查。两者互补，不能互相替代。

巡检平台可以做一些 Prometheus 不方便做的事：

  • 证书过期检查（需要主动 TLS 连接）
  • 配置文件合规性检查（需要读文件内容）
  • 多机横向对比（同一指标跨所有主机的分布）
  • 离线报告生成（不需要长期运行的监控服务）

扩展方向

这个平台的架构支持后续扩展：

  1. Agent 模式：当前是 SSH 远程执行模式，适合中小规模。超过 500 台机器后 SSH 开销显著，可以开发轻量 Agent 部署到目标机器，通过 gRPC 上报结果。

  2. Web 管理界面：当前配置通过 YAML 文件管理。可以加一层 Web UI，支持可视化配置检查项、查看历史趋势图、一键生成报告。

  3. 自动修复：巡检发现问题后，不只是告警，还可以触发自动修复脚本。比如磁盘满了自动清理日志、进程挂了自动重启。但自动修复要谨慎——错误的修复动作比问题本身更危险。

  4. 趋势分析：历史巡检数据存入数据库后，可以做趋势分析。“过去 30 天磁盘使用率的增长率是多少？按当前速度还有多少天到 95%？“这类预测性分析对容量规划很有价值。

  5. 合规检查：在检查项中加入合规规则——密码策略是否符合要求、SSH 是否禁止 root 登录、防火墙规则是否正确。把安全合规扫描整合进日常巡检。

总结

自动化巡检平台的核心价值在于：把重复的人肉巡检变成自动化的系统工程，让运维人员从"天天检查"中解放出来，把精力花在真正需要人判断的事情上。

架构设计的关键决策：

  • 插件化架构：Checker 接口 + Registry 注册表，新增检查项不改主程序，扩展性好。
  • Worker Pool 并发：用 goroutine + channel 控制并发数，200 台机器 20 并发 3 分钟跑完，比串行快 10 倍。
  • 告警去重：同一问题在抑制窗口内不重复告警，避免告警风暴。实测在 200 台机器规模下，一次磁盘满的告警从原来 50 条去重到 1 条。
  • 配置驱动：检查项、阈值、目标主机、调度计划全在 YAML 配置文件里，修改配置不需要重新编译。
  • Go 单二进制部署：一个 15MB 的二进制文件 + 一个 YAML 配置 = 完整部署，比 Python 方案省去环境管理的麻烦。

踩过的坑：

  • SSH 连接超时要设短（5 秒），否则目标机器不可达时整个巡检会被拖死。ConnectTimeout=5 加上 ServerAliveInterval=30 是比较稳的组合。
  • goroutine 不是越多越好。200 台机器开 200 个 goroutine，SSH 连接会瞬时打满目标机器的 MaxStartups。20 并发是 200 台规模下的甜点值。
  • 告警去重的时间窗口要根据业务调整。磁盘类问题 30 分钟够了，但证书过期这种每天检查一次的，去重窗口要设 24 小时，否则一天发 24 条一样的告警。
  • df 命令在不同 Linux 发行版的输出格式有差异。CentOS 7 和 Ubuntu 22.04 的 df 列顺序和格式不一样，解析时要做兼容。用 df --output=pcent,target 可以强制指定输出格式，避免这个问题。

最后一点经验：巡检平台不是替代监控系统的，而是补充。Prometheus 负责实时发现问题，巡检平台负责定期全面体检。两者结合，运维才能真正做到"心里有数”。

参考资料与致谢

本文在撰写过程中参考了以下资料，感谢原作者的贡献：

  1. 从零到一：Python 网络自动化巡检的模块化架构设计 — CSDN，模块化巡检架构设计思路，连接管理和数据处理模块的实现参考
  2. AI Agent 自动化运维体系：从手动巡检到自动驾驶的进化之路 — 腾讯云开发者社区，自动化运维体系的分层设计和模块化实践
  3. Go 并发编程实战 — Go 官方 Tour，goroutine 和 channel 的并发模型基础
  4. robfig/cron: Go cron 库 — GitHub，Go 语言的 cron 表达式解析和定时任务调度