feat: 监控告警体系统一重构 + 自动化任务调度引擎 + 监控大屏拆分

告警中心
- 新增统一告警(alert_unified)、监控告警规则(monitor_alert_rule)、指标目录(alert_metric_catalog)
- 通知策略与消息分发(alert_notify_policy/dispatch/message)、静默规则
- 告警引擎按阈值/SNMP 拆分(alert_engine_monitor/snmp)

监控采集
- SNMP 常量/模板与历史指标(snmp templates/constants、snmp_metrics_history、snmp_collect)
- 中间件监控 Redis 采集与种子(middleware_monitor_redis/seed)
- 站点探测 pkg(siteprobe) + 站点视图(site_view)

自动化任务
- cron 调度器(cronscheduler)、脚本运行器(scriptrunner)、系统任务(systask)及运行记录(cron_job_run、system_task)

审计与凭证
- 审计设置(audit_setting) + 历史清理(auditcleaner)

前端
- 监控页面(host/middleware/site/snmp)拆分为子模块并大屏化
- 告警中心多 Tab(规则/指标目录/通知策略/静默)、任务管理组件重构、Agent CLI Markdown 渲染
- 移除旧脚本卡片/详情/日志组件,统一为 ScriptTable/ScriptCodeEditor/TerminalOutput

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
LoganandClaude committed 2026-07-01 23:44:45 +08:00
1 parent 092d0e4b9e
commit 471353e3d7
190 files changed
+16640 -3484

No files matched your search

+41 -11
View File
@@ -165,17 +165,13 @@ func (a *Agent) Run(ctx context.Context, query, serverInfo string, prior []llm.M
// 裁剪历史
trimmed := a.trimMessages(messages)
// 调用 LLM(流式推送思考增量,降低长推理等待感) // Logan.Li
// ReAct 须用 Complete:Stream 无法组装 tool_calls/tool_use(GLM/Claude 等会只回文字不执行) // Logan.Li
round := a.roundNum(iter)
resp, err := a.llm.Stream(ctx, &llm.CompletionRequest{
resp, err := a.llm.Complete(ctx, &llm.CompletionRequest{
Messages: trimmed,
Tools: llmTools,
Temperature: a.config.Temperature,
MaxTokens: a.config.MaxTokens,
}, func(delta string) {
if strings.TrimSpace(delta) != "" {
a.emit(Event{Type: EventThoughtDelta, Round: round, Thought: delta})
}
})
if err != nil {
a.emit(Event{Type: EventError, Message: "LLM 调用失败: " + err.Error()})
@@ -185,15 +181,32 @@ func (a *Agent) Run(ctx context.Context, query, serverInfo string, prior []llm.M
a.totalTokens += resp.Usage.TotalTokens
a.emit(Event{Type: EventProgress, Tokens: a.totalTokens})
// 推送完整思考文本(供持久化与回放) // Logan.Li
if strings.TrimSpace(resp.Message.Content) != "" {
a.emit(Event{Type: EventThought, Round: round, Thought: resp.Message.Content})
}
messages = append(messages, resp.Message)
// 无工具调用:首轮若仅描述计划未动手,注入提示重试一次 // Logan.Li
if len(resp.Message.ToolCalls) == 0 && totalToolCalls == 0 && iter < 2 {
// 计划阶段可展示简短思考,便于用户感知进度 // Logan.Li
if thought := strings.TrimSpace(resp.Message.Content); thought != "" {
a.emit(Event{Type: EventThought, Round: round, Thought: thought})
}
messages = append(messages, llm.Message{
Role: "user",
Content: "【系统提示】请立即调用 ssh_execute 工具在目标主机执行命令获取实际结果,不要仅用文字描述计划。",
})
continue
}
// 无工具调用 → 最终答案
if len(resp.Message.ToolCalls) == 0 {
answer := resp.Message.Content
answer := strings.TrimSpace(resp.Message.Content)
// 部分模型最终轮 content 为空,回退到最近工具输出或默认文案 // Logan.Li
if answer == "" {
if fallback := lastToolOutput(messages); fallback != "" {
answer = "任务已执行完成,命令输出如下:\n" + fallback
} else {
answer = "任务已完成。"
}
}
a.emit(Event{Type: EventDone, Answer: answer, Tokens: a.totalTokens})
a.triggerEvolve(ctx, query, serverInfo, answer, toolRecords, true, start)
return &RunResult{
@@ -202,6 +215,11 @@ func (a *Agent) Run(ctx context.Context, query, serverInfo string, prior []llm.M
}, nil
}
// 中间轮思考(后续还有 tool_calls,与终局 done 分开,避免重复输出) // Logan.Li
if thought := strings.TrimSpace(resp.Message.Content); thought != "" {
a.emit(Event{Type: EventThought, Round: round, Thought: thought})
}
// 处理工具调用
observations, records, failInc := a.processToolCalls(ctx, resp.Message.ToolCalls, toolCallHistory, round)
toolRecords = append(toolRecords, records...)
@@ -346,6 +364,18 @@ func (a *Agent) roundNum(iter int) int {
return a.config.RoundOffset + iter + 1
}
// lastToolOutput 取最近一条工具 observation,供最终答案为空时回退展示 // Logan.Li
func lastToolOutput(messages []llm.Message) string {
for i := len(messages) - 1; i >= 0; i-- {
if messages[i].Role == "tool" {
if out := strings.TrimSpace(messages[i].Content); out != "" {
return out
}
}
}
return ""
}
// extractHistory 提取 system 之后的对话历史(供多轮追问) // Logan.Li
func (a *Agent) extractHistory(messages []llm.Message) []llm.Message {
if len(messages) <= 1 {
+2 -2
View File
@@ -245,7 +245,7 @@ func (c *AnthropicClient) Complete(ctx context.Context, req *CompletionRequest)
var toolCalls []ToolCall
for _, b := range parsed.Content {
switch b.Type {
case "text":
case "text", "thinking":
textBuf.WriteString(b.Text)
case "tool_use":
args := string(b.Input)
@@ -276,7 +276,7 @@ func (c *AnthropicClient) Complete(ctx context.Context, req *CompletionRequest)
// Stream 流式补全(Anthropic SSE) // Logan.Li
// 逐 token 回调 text_delta;usage 从 message_start(input) + message_delta(output) 取。
// 注:流式主要用于纯对话(智能助手),tool_use 流式增量在此不组装(Agent CLI 走 Complete)。
// 注:流式不组装 tool_use,仅用于智能助手纯对话;Agent CLI ReAct 循环必须使用 Complete。
func (c *AnthropicClient) Stream(ctx context.Context, req *CompletionRequest, onDelta func(string)) (*CompletionResponse, error) {
body, err := c.buildRequest(req, true)
if err != nil {
+79
View File
@@ -0,0 +1,79 @@
// 审计日志按保留天数清理 — 操作审计 / 登录日志 / 终端会话 / 会话命令 // Logan.Li
package auditcleaner
import (
"encoding/json"
"fmt"
"time"
"ezops/model"
"gorm.io/gorm"
)
// Result 单次清理统计 // Logan.Li
type Result struct {
RetentionDays int `json:"retention_days"`
CutoffTime time.Time `json:"cutoff_time"`
AuditLogs int64 `json:"audit_logs"`
LoginLogs int64 `json:"login_logs"`
TerminalSessions int64 `json:"terminal_sessions"`
SessionCommands int64 `json:"session_commands"`
Total int64 `json:"total"`
}
// SummaryJSON 生成可持久化的摘要 JSON // Logan.Li
func (r Result) SummaryJSON() string {
b, _ := json.Marshal(r)
return string(b)
}
// GetRetentionDays 读取当前保留天数,无配置时默认 180 天 // Logan.Li
func GetRetentionDays(db *gorm.DB) int {
var s model.AuditSettings
if err := db.First(&s).Error; err != nil || s.RetentionDays <= 0 {
return 180
}
return s.RetentionDays
}
// Run 删除 cutoff 之前的审计相关记录 // Logan.Li
func Run(db *gorm.DB, retentionDays int) (Result, error) {
if !model.IsValidRetentionDays(retentionDays) {
return Result{}, fmt.Errorf("无效的保留天数: %d", retentionDays)
}
cutoff := time.Now().AddDate(0, 0, -retentionDays)
res := Result{RetentionDays: retentionDays, CutoffTime: cutoff}
if r := db.Where("created_at < ?", cutoff).Delete(&model.SessionCommand{}); r.Error != nil {
return res, r.Error
} else {
res.SessionCommands = r.RowsAffected
}
if r := db.Where("start_time < ?", cutoff).Delete(&model.TerminalSession{}); r.Error != nil {
return res, r.Error
} else {
res.TerminalSessions = r.RowsAffected
}
if r := db.Where("login_at < ?", cutoff).Delete(&model.LoginLog{}); r.Error != nil {
return res, r.Error
} else {
res.LoginLogs = r.RowsAffected
}
if r := db.Where("created_at < ?", cutoff).Delete(&model.AuditLog{}); r.Error != nil {
return res, r.Error
} else {
res.AuditLogs = r.RowsAffected
}
res.Total = res.AuditLogs + res.LoginLogs + res.TerminalSessions + res.SessionCommands
return res, nil
}
// RunFromSettings 按库中配置执行清理 // Logan.Li
func RunFromSettings(db *gorm.DB) (Result, error) {
return Run(db, GetRetentionDays(db))
}
+138
View File
@@ -0,0 +1,138 @@
// 用户定时任务调度 — Linux 标准 5 段 crontab(分 时 日 月 周)// Logan.Li
package cronscheduler
import (
"log"
"strings"
"sync"
"time"
"ezops/model"
"github.com/robfig/cron/v3"
"gorm.io/gorm"
)
// Runner 执行定时任务回调(由 controller 注入)// Logan.Li
type Runner func(jobID uint)
type trackedEntry struct {
entryID cron.EntryID
expr string
}
// Scheduler 后台 Cron 调度器 // Logan.Li
type Scheduler struct {
db *gorm.DB
runner Runner
cron *cron.Cron
mu sync.Mutex
entries map[uint]trackedEntry
running sync.Map
}
// Start 启动调度器:加载已启用任务,并周期性同步 DB 变更 // Logan.Li
func Start(db *gorm.DB, runner Runner) *Scheduler {
s := &Scheduler{
db: db,
runner: runner,
cron: cron.New(cron.WithLocation(time.Local)),
entries: make(map[uint]trackedEntry),
}
s.cron.Start()
s.reload()
go func() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
s.reload()
}
}()
log.Println("[CronScheduler] 已启动(Linux 5 段表达式:分 时 日 月 周)")
return s
}
func (s *Scheduler) Stop() {
s.cron.Stop()
}
func (s *Scheduler) reload() {
var jobs []model.CronJob
if err := s.db.Where("enabled = ?", true).Find(&jobs).Error; err != nil {
log.Printf("[CronScheduler] 加载任务失败: %v", err)
return
}
s.mu.Lock()
defer s.mu.Unlock()
active := make(map[uint]struct{}, len(jobs))
for _, job := range jobs {
expr := normalizeCronExpr(job.CronExpr)
if expr == "" {
if prev, ok := s.entries[job.ID]; ok {
s.cron.Remove(prev.entryID)
delete(s.entries, job.ID)
}
continue
}
active[job.ID] = struct{}{}
if prev, ok := s.entries[job.ID]; ok && prev.expr == expr {
continue
}
if prev, ok := s.entries[job.ID]; ok {
s.cron.Remove(prev.entryID)
}
jobID := job.ID
entryID, err := s.cron.AddFunc(expr, func() {
s.dispatch(jobID)
})
if err != nil {
log.Printf("[CronScheduler] 注册任务 %s(%d) 失败 expr=%q: %v", job.Name, job.ID, expr, err)
delete(s.entries, job.ID)
continue
}
s.entries[job.ID] = trackedEntry{entryID: entryID, expr: expr}
log.Printf("[CronScheduler] 已注册: %s id=%d expr=%s", job.Name, job.ID, expr)
}
for id, prev := range s.entries {
if _, ok := active[id]; !ok {
s.cron.Remove(prev.entryID)
delete(s.entries, id)
log.Printf("[CronScheduler] 已移除: id=%d", id)
}
}
}
func (s *Scheduler) dispatch(jobID uint) {
if _, loaded := s.running.LoadOrStore(jobID, true); loaded {
log.Printf("[CronScheduler] 任务 id=%d 仍在执行,跳过本次调度", jobID)
return
}
defer s.running.Delete(jobID)
if s.runner == nil {
return
}
s.runner(jobID)
}
// normalizeCronExpr 规整为 Linux crontab 五段式 // Logan.Li
func normalizeCronExpr(expr string) string {
expr = strings.TrimSpace(expr)
if expr == "" {
return ""
}
fields := strings.Fields(expr)
if len(fields) == 5 {
return strings.Join(fields, " ")
}
// 误填 6 段(含秒)时去掉首段,兼容常见 Quartz 写法 // Logan.Li
if len(fields) == 6 {
return strings.Join(fields[1:], " ")
}
return ""
}
+13 -2
View File
@@ -91,6 +91,10 @@ func AutoMigrate(db *gorm.DB) error {
&model.KubernetesIngress{},
&model.AlertRule{},
&model.AlertEvent{},
&model.AlertContact{}, // 报警联系人 // Logan.Li
&model.AlertContactGroup{}, // 报警联系组 // Logan.Li
&model.AlertNotifyPolicy{}, // 告警集合通知策略 // Logan.Li
&model.AlertNotifySettings{}, // 告警消息模版 // Logan.Li
&model.NotifyChannel{},
&model.MetricData{},
&model.Pipeline{},
@@ -105,6 +109,9 @@ func AutoMigrate(db *gorm.DB) error {
&model.NavigationGroup{},
&model.NavigationLink{},
&model.AuditLog{},
&model.AuditSettings{}, // 审计保留策略 // Logan.Li
&model.LoginLog{}, // 登录日志 // Logan.Li
&model.SystemTaskStatus{}, // 内置系统任务状态 // Logan.Li
&model.LDAPConfig{},
&model.ResourceModel{},
&model.CloudBilling{},
@@ -129,11 +136,15 @@ func AutoMigrate(db *gorm.DB) error {
&model.MetricValue{}, // 指标当前值 // Logan.Li
&model.MetricHistory{}, // 指标历史时序 // Logan.Li
&model.HostGroup{}, // 主机组 // Logan.Li
&model.SNMPDevice{}, // SNMP设备 // Logan.Li
&model.SNMPTemplate{}, // SNMP模板 // Logan.Li
&model.SNMPDevice{}, // SNMP设备 // Logan.Li
&model.SNMPTemplate{}, // SNMP模板 // Logan.Li
&model.SNMPTemplateOID{}, // SNMP模板OID // Logan.Li
&model.SNMPMetric{}, // SNMP采集指标 // Logan.Li
&model.SNMPInterface{}, // SNMP接口信息 // Logan.Li
&model.SiteTask{}, // 站点拨测任务 // Logan.Li
&model.SiteResult{}, // 拨测结果 // Logan.Li
&model.SiteAlert{}, // 拨测告警 // Logan.Li
&model.MonitorAlertRule{}, // 中间件/站点拨测告警规则 // Logan.Li
&model.PrometheusDataSource{}, // Prometheus 数据源配置 // Logan.Li
&model.ZabbixDataSource{}, // Zabbix 数据源配置 // Logan.Li
&model.LogDataSource{}, // Loki / ES 日志数据源 // Logan.Li
+20
View File
@@ -1,6 +1,7 @@
package database
import (
"fmt"
"strings"
"gorm.io/gorm"
@@ -40,4 +41,23 @@ func CleanupAgentReportsSQL(driver string) string {
default:
return "DELETE FROM monitor_agent_reports WHERE report_time < DATE_SUB(NOW(), INTERVAL 7 DAY)"
}
}
// CleanupSNMPMetricsSQL 清理超期 SNMP 采集指标 // Logan.Li
func CleanupSNMPMetricsSQL(driver string, retentionDays int) string {
if retentionDays <= 0 {
retentionDays = 7
}
switch driver {
case "sqlite":
return fmt.Sprintf(
"DELETE FROM ops_snmp_metrics WHERE timestamp < CAST(strftime('%%s', 'now', '-%d days') AS INTEGER)",
retentionDays,
)
default:
return fmt.Sprintf(
"DELETE FROM ops_snmp_metrics WHERE timestamp < UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL %d DAY))",
retentionDays,
)
}
}
+159
View File
@@ -0,0 +1,159 @@
// 脚本运行时探测与本地执行 — 适配 ./aiops 单二进制 / Docker 容器环境 // Logan.Li
package scriptrunner
import (
"bytes"
"context"
"fmt"
"os"
"os/exec"
"runtime"
"time"
)
// RuntimeInfo 当前主机可用的脚本解释器
type RuntimeInfo struct {
Shell bool `json:"shell"`
ShellPath string `json:"shell_path,omitempty"`
Python bool `json:"python"`
PythonPath string `json:"python_path,omitempty"`
PowerShell bool `json:"powershell"`
PowerShellPath string `json:"powershell_path,omitempty"`
OS string `json:"os"`
Arch string `json:"arch"`
EmbedMode bool `json:"embed_mode"` // 单二进制 embed 部署
}
// DetectRuntime 探测 bash/sh、python3、powershell 是否可用 // Logan.Li
func DetectRuntime() RuntimeInfo {
info := RuntimeInfo{
OS: runtime.GOOS,
Arch: runtime.GOARCH,
EmbedMode: true,
}
for _, p := range []string{"/bin/bash", "/usr/bin/bash", "bash"} {
if path, ok := lookExecutable(p); ok {
info.Shell = true
info.ShellPath = path
break
}
}
if !info.Shell {
for _, p := range []string{"/bin/sh", "/usr/bin/sh", "sh"} {
if path, ok := lookExecutable(p); ok {
info.Shell = true
info.ShellPath = path
break
}
}
}
for _, p := range []string{"/usr/bin/python3", "/usr/local/bin/python3", "python3"} {
if path, ok := lookExecutable(p); ok {
info.Python = true
info.PythonPath = path
break
}
}
for _, p := range []string{"/usr/bin/pwsh", "/usr/local/bin/pwsh", "pwsh", "powershell"} {
if path, ok := lookExecutable(p); ok {
info.PowerShell = true
info.PowerShellPath = path
break
}
}
return info
}
func lookExecutable(name string) (string, bool) {
path, err := exec.LookPath(name)
if err != nil {
return "", false
}
return path, true
}
// CanRun 判断指定脚本类型是否可在本机执行 // Logan.Li
func CanRun(rt RuntimeInfo, scriptType string) (bool, string) {
switch scriptType {
case "shell", "bash", "command", "":
if rt.Shell {
return true, ""
}
return false, "当前环境未找到 bash/sh,无法执行 Shell 脚本(Docker 镜像需安装 bash)"
case "python":
if rt.Python {
return true, ""
}
return false, "当前环境未找到 python3,无法执行 Python 脚本(请在镜像中安装 python3)"
case "powershell":
if rt.PowerShell {
return true, ""
}
return false, "当前环境未找到 PowerShell (pwsh),无法执行 PowerShell 脚本"
default:
return false, fmt.Sprintf("不支持的脚本类型: %s", scriptType)
}
}
// Run 在 AIOps 本机执行脚本,返回合并输出、退出码、耗时(ms) // Logan.Li
func Run(scriptType, content string, timeoutSec int) (output string, exitCode int, durationMs int, err error) {
rt := DetectRuntime()
ok, reason := CanRun(rt, scriptType)
if !ok {
return reason + "\n", -1, 0, fmt.Errorf("%s", reason)
}
start := time.Now()
timeout := time.Duration(timeoutSec) * time.Second
if timeout <= 0 {
timeout = 300 * time.Second
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
var cmd *exec.Cmd
switch scriptType {
case "python":
py := rt.PythonPath
if py == "" {
py = "python3"
}
cmd = exec.CommandContext(ctx, py, "-c", content)
case "powershell":
pwsh := rt.PowerShellPath
if pwsh == "" {
pwsh = "pwsh"
}
cmd = exec.CommandContext(ctx, pwsh, "-NoProfile", "-NonInteractive", "-Command", content)
default:
sh := rt.ShellPath
if sh == "" {
sh = "bash"
}
cmd = exec.CommandContext(ctx, sh, "-c", content)
}
var out bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &out
cmd.Env = os.Environ()
runErr := cmd.Run()
durationMs = int(time.Since(start).Milliseconds())
output = out.String()
if ctx.Err() == context.DeadlineExceeded {
output += fmt.Sprintf("\n[超时] 执行超过 %d 秒被终止", timeoutSec)
return output, -1, durationMs, ctx.Err()
}
if runErr != nil {
if exitError, ok := runErr.(*exec.ExitError); ok {
exitCode = exitError.ExitCode()
} else {
exitCode = -1
output += "\n" + runErr.Error()
}
return output, exitCode, durationMs, runErr
}
return output, 0, durationMs, nil
}
+277
View File
@@ -0,0 +1,277 @@
package siteprobe
import (
"context"
"crypto/tls"
"fmt"
"net"
"net/http"
"os/exec"
"regexp"
"strconv"
"strings"
"time"
"ezops/model"
)
// ProbeOutcome 单次拨测结果 // Logan.Li
type ProbeOutcome struct {
Status model.SiteResultStatus
HTTPStatus int
ResponseTime float64
TotalTime float64
RemoteIP string
ErrorMessage string
// HTTPS 证书信息 // Logan.Li
SSLCertExpiry *time.Time
SSLCertIssuer string
SSLCertSubject string
SSLDaysRemaining int
}
// Execute 按任务类型执行拨测 // Logan.Li
func Execute(ctx context.Context, task model.SiteTask) ProbeOutcome {
timeout := time.Duration(task.Timeout) * time.Second
if timeout <= 0 {
timeout = 10 * time.Second
}
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
taskType := strings.ToLower(string(task.TaskType))
if taskType == "" {
taskType = inferTaskType(task)
}
switch taskType {
case "ssl":
return probeSSL(ctx, task)
case "tcp":
return probeTCP(ctx, task.URL)
case "dns":
return probeDNS(ctx, task.URL, task.DNSResolver)
case "ping":
return probePing(ctx, task.URL, task.Timeout)
case "http", "https":
return probeHTTP(ctx, task)
default:
return probeHTTP(ctx, task)
}
}
// InferTaskTypePublic 推断任务类型(供列表修复存量数据)// Logan.Li
func InferTaskTypePublic(task model.SiteTask) string {
return inferTaskType(task)
}
func inferTaskType(task model.SiteTask) string {
name := strings.ToLower(task.Name)
switch {
case strings.Contains(name, "tcp"):
return "tcp"
case strings.Contains(name, "dns"):
return "dns"
case strings.Contains(name, "ping"):
return "ping"
case strings.Contains(name, "ssl"):
return "ssl"
default:
return "http"
}
}
func probeHTTP(ctx context.Context, task model.SiteTask) ProbeOutcome {
start := time.Now()
url := task.URL
if !strings.HasPrefix(url, "http://") && !strings.HasPrefix(url, "https://") {
url = "http://" + url
}
transport := &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: !task.VerifySSL},
}
client := &http.Client{
Timeout: time.Until(deadlineFromCtx(ctx)),
Transport: transport,
}
if !task.FollowRedirect {
client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
}
}
method := strings.ToUpper(string(task.Method))
if method == "" {
method = http.MethodGet
}
req, err := http.NewRequestWithContext(ctx, method, url, nil)
if err != nil {
return failOutcome(start, err.Error())
}
resp, err := client.Do(req)
elapsed := msSince(start)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return ProbeOutcome{Status: model.SiteResultTimeout, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: "请求超时"}
}
return failOutcome(start, err.Error())
}
defer resp.Body.Close()
statusOK := resp.StatusCode == task.ExpectedStatus
if task.ExpectedStatus == 0 {
statusOK = resp.StatusCode >= 200 && resp.StatusCode < 400
}
if !statusOK {
return ProbeOutcome{
Status: model.SiteResultFail,
HTTPStatus: resp.StatusCode,
ResponseTime: elapsed,
TotalTime: elapsed,
ErrorMessage: fmt.Sprintf("状态码 %d 不符合期望 %d", resp.StatusCode, task.ExpectedStatus),
}
}
return ProbeOutcome{
Status: model.SiteResultSuccess,
HTTPStatus: resp.StatusCode,
ResponseTime: elapsed,
TotalTime: elapsed,
}
}
func probeTCP(ctx context.Context, target string) ProbeOutcome {
start := time.Now()
host, port, err := splitHostPort(target, "80")
if err != nil {
return failOutcome(start, err.Error())
}
d := net.Dialer{Timeout: time.Until(deadlineFromCtx(ctx))}
conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(host, port))
elapsed := msSince(start)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return ProbeOutcome{Status: model.SiteResultTimeout, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: "TCP 连接超时"}
}
return failOutcome(start, err.Error())
}
defer conn.Close()
remote := ""
if addr := conn.RemoteAddr(); addr != nil {
remote = addr.String()
}
return ProbeOutcome{
Status: model.SiteResultSuccess,
ResponseTime: elapsed,
TotalTime: elapsed,
RemoteIP: remote,
}
}
func probeDNS(ctx context.Context, target, resolver string) ProbeOutcome {
start := time.Now()
domain := normalizeDomain(target)
if domain == "" {
return failOutcome(start, "域名不能为空")
}
r := &net.Resolver{}
if resolver != "" {
r = &net.Resolver{
PreferGo: true,
Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
d := net.Dialer{Timeout: time.Until(deadlineFromCtx(ctx))}
return d.DialContext(ctx, network, resolver+":53")
},
}
}
addrs, err := r.LookupHost(ctx, domain)
elapsed := msSince(start)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return ProbeOutcome{Status: model.SiteResultTimeout, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: "DNS 解析超时"}
}
return failOutcome(start, err.Error())
}
return ProbeOutcome{
Status: model.SiteResultSuccess,
ResponseTime: elapsed,
TotalTime: elapsed,
RemoteIP: strings.Join(addrs, ","),
}
}
func probePing(ctx context.Context, target string, timeoutSec int) ProbeOutcome {
start := time.Now()
host := normalizeDomain(target)
if host == "" {
return failOutcome(start, "主机不能为空")
}
if timeoutSec <= 0 {
timeoutSec = 5
}
cmd := exec.CommandContext(ctx, "ping", "-c", "1", "-W", strconv.Itoa(timeoutSec), host)
out, err := cmd.CombinedOutput()
elapsed := msSince(start)
if err != nil {
msg := strings.TrimSpace(string(out))
if msg == "" {
msg = err.Error()
}
if ctx.Err() == context.DeadlineExceeded {
return ProbeOutcome{Status: model.SiteResultTimeout, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: "PING 超时"}
}
return ProbeOutcome{Status: model.SiteResultFail, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: msg}
}
if m := regexp.MustCompile(`time[=<]([\d.]+)\s*ms`).FindStringSubmatch(string(out)); len(m) == 2 {
if v, parseErr := strconv.ParseFloat(m[1], 64); parseErr == nil {
elapsed = v
}
}
return ProbeOutcome{Status: model.SiteResultSuccess, ResponseTime: elapsed, TotalTime: elapsed, RemoteIP: host}
}
func splitHostPort(raw, defaultPort string) (string, string, error) {
raw = strings.TrimSpace(raw)
for _, prefix := range []string{"ssl://", "tcp://", "http://", "https://"} {
raw = strings.TrimPrefix(raw, prefix)
}
if host, port, err := net.SplitHostPort(raw); err == nil {
return host, port, nil
}
if raw == "" {
return "", "", fmt.Errorf("TCP 地址不能为空")
}
return raw, defaultPort, nil
}
func normalizeDomain(raw string) string {
raw = strings.TrimSpace(raw)
raw = strings.TrimPrefix(raw, "http://")
raw = strings.TrimPrefix(raw, "https://")
if host, _, err := net.SplitHostPort(raw); err == nil {
return host
}
return strings.TrimSuffix(raw, "/")
}
func failOutcome(start time.Time, msg string) ProbeOutcome {
elapsed := msSince(start)
return ProbeOutcome{
Status: model.SiteResultFail,
ResponseTime: elapsed,
TotalTime: elapsed,
ErrorMessage: msg,
}
}
func msSince(start time.Time) float64 {
return float64(time.Since(start).Milliseconds())
}
func deadlineFromCtx(ctx context.Context) time.Time {
if d, ok := ctx.Deadline(); ok {
return d
}
return time.Now().Add(10 * time.Second)
}
+108
View File
@@ -0,0 +1,108 @@
package siteprobe
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"net"
"net/url"
"strings"
"time"
"ezops/model"
)
// parseSSLTarget 解析 SSL 拨测地址,支持 example.com:443 / https://example.com // Logan.Li
func parseSSLTarget(raw string) (host, port string, err error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return "", "", fmt.Errorf("SSL 地址不能为空")
}
lower := strings.ToLower(raw)
// 兼容 https:example.com(单斜杠写法)
if strings.HasPrefix(lower, "https:") && !strings.HasPrefix(lower, "https://") {
raw = "https://" + strings.TrimSpace(raw[len("https:"):])
lower = strings.ToLower(raw)
}
if strings.HasPrefix(lower, "http://") || strings.HasPrefix(lower, "https://") || strings.HasPrefix(lower, "ssl://") {
u, parseErr := url.Parse(raw)
if parseErr != nil || u.Host == "" {
return "", "", fmt.Errorf("SSL 地址格式无效: %s", raw)
}
if h, p, splitErr := net.SplitHostPort(u.Host); splitErr == nil {
return h, p, nil
}
defPort := "443"
if u.Scheme == "http" {
defPort = "80"
}
return u.Hostname(), defPort, nil
}
// 去掉路径后缀,如 example.com:443/foo
if idx := strings.Index(raw, "/"); idx >= 0 {
raw = raw[:idx]
}
return splitHostPort(raw, "443")
}
// probeSSL 纯 TLS 握手检测证书(不发起 HTTP 请求)// Logan.Li
func probeSSL(ctx context.Context, task model.SiteTask) ProbeOutcome {
start := time.Now()
host, port, err := parseSSLTarget(task.URL)
if err != nil {
return failOutcome(start, err.Error())
}
addr := net.JoinHostPort(host, port)
d := &net.Dialer{Timeout: time.Until(deadlineFromCtx(ctx))}
conn, err := tls.DialWithDialer(d, "tcp", addr, &tls.Config{
ServerName: host,
InsecureSkipVerify: !task.VerifySSL,
})
elapsed := msSince(start)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return ProbeOutcome{Status: model.SiteResultTimeout, TotalTime: elapsed, ResponseTime: elapsed, ErrorMessage: "TLS 握手超时"}
}
return failOutcome(start, fmt.Sprintf("TLS 握手失败: %s", err.Error()))
}
defer conn.Close()
remote := ""
if addr := conn.RemoteAddr(); addr != nil {
remote = addr.String()
}
out := ProbeOutcome{
Status: model.SiteResultSuccess,
ResponseTime: elapsed,
TotalTime: elapsed,
RemoteIP: remote,
}
applyCertsToOutcome(&out, conn.ConnectionState().PeerCertificates)
return out
}
// applyCertsToOutcome 从 TLS 证书链填充到期信息 // Logan.Li
func applyCertsToOutcome(out *ProbeOutcome, certs []*x509.Certificate) {
if len(certs) == 0 {
return
}
cert := certs[0]
expiry := cert.NotAfter
days := int(time.Until(expiry).Hours() / 24)
out.SSLCertExpiry = &expiry
out.SSLDaysRemaining = days
out.SSLCertIssuer = cert.Issuer.CommonName
if out.SSLCertIssuer == "" {
out.SSLCertIssuer = cert.Issuer.String()
}
out.SSLCertSubject = cert.Subject.CommonName
if out.SSLCertSubject == "" {
out.SSLCertSubject = cert.Subject.String()
}
}
// IsSSLProbeTask 是否为 SSL 证书拨测任务 // Logan.Li
func IsSSLProbeTask(task model.SiteTask) bool {
return strings.EqualFold(string(task.TaskType), "ssl")
}
+37
View File
@@ -0,0 +1,37 @@
package siteprobe
import "testing"
// SSL 地址解析单测 // Logan.Li
func TestParseSSLTarget(t *testing.T) {
cases := []struct {
raw string
host string
port string
wantFail bool
}{
{"example.com:443", "example.com", "443", false},
{"example.com", "example.com", "443", false},
{"https://example.com", "example.com", "443", false},
{"https://example.com:8443", "example.com", "8443", false},
{"https://example.com/path", "example.com", "443", false},
{"https:example.com", "example.com", "443", false},
{"ssl://example.com:443", "example.com", "443", false},
{"", "", "", true},
}
for _, c := range cases {
host, port, err := parseSSLTarget(c.raw)
if c.wantFail {
if err == nil {
t.Fatalf("expected error for %q", c.raw)
}
continue
}
if err != nil {
t.Fatalf("parseSSLTarget(%q): %v", c.raw, err)
}
if host != c.host || port != c.port {
t.Fatalf("parseSSLTarget(%q) = %s:%s, want %s:%s", c.raw, host, port, c.host, c.port)
}
}
}
+118
View File
@@ -0,0 +1,118 @@
package siteprobe
import (
"context"
"log"
"sync"
"time"
"ezops/model"
"gorm.io/gorm"
)
// Worker 站点拨测后台采集器 // Logan.Li
type Worker struct {
db *gorm.DB
stopCh chan struct{}
wg sync.WaitGroup
}
// NewWorker 创建拨测 worker
func NewWorker(db *gorm.DB) *Worker {
return &Worker{db: db, stopCh: make(chan struct{})}
}
// Start 启动拨测循环
func (w *Worker) Start(intervalSec int) {
if intervalSec < 5 {
intervalSec = 10
}
w.wg.Add(1)
go func() {
defer w.wg.Done()
ticker := time.NewTicker(time.Duration(intervalSec) * time.Second)
defer ticker.Stop()
log.Printf("[SiteProbe] 拨测引擎启动,扫描间隔 %ds", intervalSec)
w.runOnce()
for {
select {
case <-ticker.C:
w.runOnce()
case <-w.stopCh:
log.Printf("[SiteProbe] 拨测引擎停止")
return
}
}
}()
}
// Stop 停止拨测
func (w *Worker) Stop() {
close(w.stopCh)
w.wg.Wait()
}
// RunTaskNow 立即执行单个任务拨测 // Logan.Li
func (w *Worker) RunTaskNow(taskID uint) error {
var task model.SiteTask
if err := w.db.First(&task, taskID).Error; err != nil {
return err
}
w.probeTask(task)
return nil
}
func (w *Worker) runOnce() {
var tasks []model.SiteTask
w.db.Where("status = ?", model.SiteTaskStatusRunning).Find(&tasks)
now := time.Now()
for _, task := range tasks {
if task.Frequency <= 0 {
task.Frequency = 60
}
if task.LastCheckAt != nil && now.Sub(*task.LastCheckAt) < time.Duration(task.Frequency)*time.Second {
continue
}
w.probeTask(task)
}
}
func (w *Worker) probeTask(task model.SiteTask) {
if task.TaskType == "" {
task.TaskType = model.SiteTaskType(inferTaskType(task))
}
outcome := Execute(context.Background(), task)
now := time.Now()
result := model.SiteResult{
TaskID: task.ID,
Status: outcome.Status,
HTTPStatusCode: outcome.HTTPStatus,
ResponseTime: outcome.ResponseTime,
TotalTime: outcome.TotalTime,
RemoteIP: outcome.RemoteIP,
ErrorMessage: outcome.ErrorMessage,
SSLCertExpiry: outcome.SSLCertExpiry,
SSLCertIssuer: outcome.SSLCertIssuer,
SSLCertSubject: outcome.SSLCertSubject,
SSLDaysRemaining: outcome.SSLDaysRemaining,
CheckedAt: now,
}
w.db.Create(&result)
lastResult := "fail"
if outcome.Status == model.SiteResultSuccess {
lastResult = "success"
}
avg := outcome.ResponseTime
if task.AvgResponse > 0 {
avg = (task.AvgResponse + outcome.ResponseTime) / 2
}
w.db.Model(&model.SiteTask{}).Where("id = ?", task.ID).Updates(map[string]interface{}{
"task_type": task.TaskType,
"last_result": lastResult,
"last_check_at": now,
"avg_response": avg,
})
}
+8
View File
@@ -0,0 +1,8 @@
package snmp
// SNMP 采集与保留策略常量 // Logan.Li
const (
DefaultCollectIntervalSec = 300 // 默认采集间隔 5 分钟
MinCollectIntervalSec = 300 // 最低采集间隔,避免高频轮询
MetricsRetentionDays = 7 // 指标保留天数
)
+130
View File
@@ -0,0 +1,130 @@
package snmp
// TemplateOIDDef 内置模板 OID 定义 // Logan.Li
type TemplateOIDDef struct {
Name string
OID string
Description string
DataType string
Unit string
IsTable bool
SortOrder int
}
// BuiltinTemplate 内置 SNMP 模板 // Logan.Li
type BuiltinTemplate struct {
Name string
Description string
DeviceType string
Vendor string
OIDs []TemplateOIDDef
}
// mergeOIDs 合并多组 OID 定义 // Logan.Li
func mergeOIDs(groups ...[]TemplateOIDDef) []TemplateOIDDef {
out := make([]TemplateOIDDef, 0)
for _, g := range groups {
out = append(out, g...)
}
return out
}
// commonNetworkOIDs 网络设备通用 IF-MIB 指标 // Logan.Li
var commonNetworkOIDs = []TemplateOIDDef{
{Name: "sys_uptime", OID: "1.3.6.1.2.1.1.3.0", Description: "系统运行时间", DataType: "timeticks", Unit: "ticks", SortOrder: 1},
{Name: "sys_name", OID: "1.3.6.1.2.1.1.5.0", Description: "系统名称", DataType: "string", SortOrder: 2},
{Name: "if_in_octets", OID: "1.3.6.1.2.1.31.1.1.1.6", Description: "接口入流量(64位)", DataType: "counter64", Unit: "bytes", IsTable: true, SortOrder: 10},
{Name: "if_out_octets", OID: "1.3.6.1.2.1.31.1.1.1.10", Description: "接口出流量(64位)", DataType: "counter64", Unit: "bytes", IsTable: true, SortOrder: 11},
{Name: "if_oper_status", OID: "1.3.6.1.2.1.2.2.1.8", Description: "接口运行状态", DataType: "integer", IsTable: true, SortOrder: 12},
{Name: "if_name", OID: "1.3.6.1.2.1.31.1.1.1.1", Description: "接口名称", DataType: "string", IsTable: true, SortOrder: 13},
{Name: "if_high_speed", OID: "1.3.6.1.2.1.31.1.1.1.15", Description: "接口速率", DataType: "gauge32", Unit: "Mbps", IsTable: true, SortOrder: 14},
}
// serverCommonOIDs 服务器 BMC 通用指标 // Logan.Li
var serverCommonOIDs = []TemplateOIDDef{
{Name: "sys_uptime", OID: "1.3.6.1.2.1.1.3.0", Description: "系统运行时间", DataType: "timeticks", Unit: "ticks", SortOrder: 1},
{Name: "sys_name", OID: "1.3.6.1.2.1.1.5.0", Description: "系统名称", DataType: "string", SortOrder: 2},
{Name: "if_in_octets", OID: "1.3.6.1.2.1.31.1.1.1.6", Description: "带外网卡入流量", DataType: "counter64", Unit: "bytes", IsTable: true, SortOrder: 10},
{Name: "if_out_octets", OID: "1.3.6.1.2.1.31.1.1.1.10", Description: "带外网卡出流量", DataType: "counter64", Unit: "bytes", IsTable: true, SortOrder: 11},
{Name: "ent_sensor_value", OID: "1.3.6.1.2.1.99.1.1.1.4", Description: "实体传感器值(温度/风扇兜底)", DataType: "gauge32", IsTable: true, SortOrder: 15},
{Name: "hr_storage_size", OID: "1.3.6.1.2.1.25.2.3.1.5", Description: "存储容量", DataType: "integer", Unit: "KB", IsTable: true, SortOrder: 16},
{Name: "hr_storage_used", OID: "1.3.6.1.2.1.25.2.3.1.6", Description: "存储已用", DataType: "integer", Unit: "KB", IsTable: true, SortOrder: 17},
}
// h3cOIDs H3C Comware 私有指标 // Logan.Li
var h3cOIDs = []TemplateOIDDef{
{Name: "h3c_cpu_usage", OID: "1.3.6.1.4.1.25506.2.6.1.1.1.1.6", Description: "CPU 使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 20},
{Name: "h3c_mem_usage", OID: "1.3.6.1.4.1.25506.2.6.1.1.1.1.8", Description: "内存使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 21},
{Name: "h3c_temperature", OID: "1.3.6.1.4.1.25506.2.6.1.1.1.1.12", Description: "设备温度", DataType: "gauge32", Unit: "°C", IsTable: true, SortOrder: 22},
{Name: "h3c_fan_speed", OID: "1.3.6.1.4.1.25506.2.6.1.1.1.1.14", Description: "风扇转速", DataType: "gauge32", Unit: "RPM", IsTable: true, SortOrder: 23},
}
// huaweiOIDs 华为 VRP 私有指标 // Logan.Li
var huaweiOIDs = []TemplateOIDDef{
{Name: "hw_cpu_usage", OID: "1.3.6.1.4.1.2011.5.25.31.1.1.1.1.5", Description: "CPU 使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 20},
{Name: "hw_mem_usage", OID: "1.3.6.1.4.1.2011.5.25.31.1.1.1.1.7", Description: "内存使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 21},
{Name: "hw_temperature", OID: "1.3.6.1.4.1.2011.5.25.31.1.1.1.1.11", Description: "设备温度", DataType: "gauge32", Unit: "°C", IsTable: true, SortOrder: 22},
{Name: "hw_fan_speed", OID: "1.3.6.1.4.1.2011.5.25.31.1.1.1.1.12", Description: "风扇转速", DataType: "gauge32", Unit: "RPM", IsTable: true, SortOrder: 23},
}
// ciscoOIDs 思科 IOS/NX-OS 私有指标 // Logan.Li
var ciscoOIDs = []TemplateOIDDef{
{Name: "cisco_cpu_5min", OID: "1.3.6.1.4.1.9.9.109.1.1.1.1.8", Description: "CPU 5分钟利用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 20},
{Name: "cisco_mem_used", OID: "1.3.6.1.4.1.9.9.48.1.1.1.5", Description: "内存已用", DataType: "gauge32", Unit: "bytes", IsTable: true, SortOrder: 21},
{Name: "cisco_mem_free", OID: "1.3.6.1.4.1.9.9.48.1.1.1.6", Description: "内存空闲", DataType: "gauge32", Unit: "bytes", IsTable: true, SortOrder: 22},
{Name: "cisco_sensor_value", OID: "1.3.6.1.4.1.9.9.91.1.1.1.4", Description: "温度/风扇传感器", DataType: "gauge32", IsTable: true, SortOrder: 23},
}
// ruijieOIDs 锐捷 RGOS 私有指标 // Logan.Li
var ruijieOIDs = []TemplateOIDDef{
{Name: "ruijie_cpu_usage", OID: "1.3.6.1.4.1.4881.1.1.10.2.1.1.1.1.6", Description: "CPU 使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 20},
{Name: "ruijie_mem_usage", OID: "1.3.6.1.4.1.4881.1.1.10.2.1.2.1.1.6", Description: "内存使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 21},
{Name: "ruijie_temperature", OID: "1.3.6.1.4.1.4881.1.1.10.2.1.3.1.1.6", Description: "设备温度", DataType: "gauge32", Unit: "°C", IsTable: true, SortOrder: 22},
{Name: "ruijie_fan_speed", OID: "1.3.6.1.4.1.4881.1.1.10.2.1.4.1.1.6", Description: "风扇转速", DataType: "gauge32", Unit: "RPM", IsTable: true, SortOrder: 23},
}
// dellIdracOIDs Dell iDRAC 带外管理指标 // Logan.Li
var dellIdracOIDs = []TemplateOIDDef{
{Name: "dell_global_status", OID: "1.3.6.1.4.1.674.10892.5.4.1.0", Description: "整机健康状态", DataType: "integer", SortOrder: 20},
{Name: "dell_cpu_status", OID: "1.3.6.1.4.1.674.10892.5.4.701.12.6.1.8", Description: "CPU 状态", DataType: "integer", IsTable: true, SortOrder: 21},
{Name: "dell_temperature", OID: "1.3.6.1.4.1.674.10892.5.4.700.20.1.6.1.8", Description: "温度探头读数", DataType: "gauge32", Unit: "°C", IsTable: true, SortOrder: 22},
{Name: "dell_fan_speed", OID: "1.3.6.1.4.1.674.10892.5.4.700.20.1.7.1.8", Description: "散热风扇读数", DataType: "gauge32", Unit: "RPM", IsTable: true, SortOrder: 23},
{Name: "dell_disk_state", OID: "1.3.6.1.4.1.674.10892.5.5.1300.20.1.4.1.4", Description: "物理磁盘状态", DataType: "integer", IsTable: true, SortOrder: 24},
{Name: "dell_disk_capacity", OID: "1.3.6.1.4.1.674.10892.5.5.1300.20.1.4.1.11", Description: "物理磁盘容量", DataType: "gauge32", Unit: "MB", IsTable: true, SortOrder: 25},
}
// inspurBmcOIDs 浪潮 BMC 私有指标 // Logan.Li
var inspurBmcOIDs = []TemplateOIDDef{
{Name: "inspur_cpu_usage", OID: "1.3.6.1.4.1.37945.1.1.1.1.1.1", Description: "CPU 使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 20},
{Name: "inspur_mem_usage", OID: "1.3.6.1.4.1.37945.1.1.1.1.2.1", Description: "内存使用率", DataType: "gauge32", Unit: "%", IsTable: true, SortOrder: 21},
{Name: "inspur_temperature", OID: "1.3.6.1.4.1.37945.1.1.1.1.3.1", Description: "温度传感器", DataType: "gauge32", Unit: "°C", IsTable: true, SortOrder: 22},
{Name: "inspur_fan_speed", OID: "1.3.6.1.4.1.37945.1.1.1.1.4.1", Description: "风扇转速", DataType: "gauge32", Unit: "RPM", IsTable: true, SortOrder: 23},
{Name: "inspur_disk_status", OID: "1.3.6.1.4.1.37945.1.1.1.1.5.1", Description: "磁盘健康状态", DataType: "integer", IsTable: true, SortOrder: 24},
}
// BuiltinTemplates 全部内置模板 // Logan.Li
func BuiltinTemplates() []BuiltinTemplate {
net := func(vendor string, deviceType string, name string, desc string, priv []TemplateOIDDef) BuiltinTemplate {
return BuiltinTemplate{
Name: name, Description: desc, DeviceType: deviceType, Vendor: vendor,
OIDs: mergeOIDs(commonNetworkOIDs, priv),
}
}
srv := func(vendor string, name string, desc string, priv []TemplateOIDDef) BuiltinTemplate {
return BuiltinTemplate{
Name: name, Description: desc, DeviceType: "server", Vendor: vendor,
OIDs: mergeOIDs(serverCommonOIDs, priv),
}
}
return []BuiltinTemplate{
net("H3C", "switch", "H3C 交换机", "H3C Comware 交换机 — CPU/内存/温度/风扇/网卡流量", h3cOIDs),
net("H3C", "router", "H3C 路由器", "H3C Comware 路由器 — CPU/内存/温度/风扇/网卡流量", h3cOIDs),
net("Huawei", "switch", "华为交换机", "华为 VRP 交换机 — CPU/内存/温度/风扇/网卡流量", huaweiOIDs),
net("Huawei", "router", "华为路由器", "华为 VRP 路由器 — CPU/内存/温度/风扇/网卡流量", huaweiOIDs),
net("Cisco", "switch", "思科交换机", "Cisco IOS/NX-OS — CPU/内存/传感器/网卡流量", ciscoOIDs),
net("Cisco", "router", "思科路由器", "Cisco IOS — CPU/内存/传感器/网卡流量", ciscoOIDs),
net("Ruijie", "switch", "锐捷交换机", "锐捷 RGOS — CPU/内存/温度/风扇/网卡流量", ruijieOIDs),
srv("Dell", "Dell iDRAC", "Dell 服务器 iDRAC — CPU/温度/风扇/磁盘/带外网卡", dellIdracOIDs),
srv("Inspur", "浪潮 BMC", "浪潮服务器 BMC — CPU/内存/温度/风扇/磁盘/带外网卡", inspurBmcOIDs),
}
}
+226
View File
@@ -0,0 +1,226 @@
// 内置系统定时任务注册表 — 告警引擎 / 数据清理等 // Logan.Li
package systask
import (
"fmt"
"sync"
"time"
"ezops/model"
"ezops/pkg/auditcleaner"
"ezops/pkg/database"
snmppkg "ezops/pkg/snmp"
"gorm.io/gorm"
)
// Def 系统任务元数据 // Logan.Li
type Def struct {
Code string `json:"code"`
Name string `json:"name"`
Description string `json:"description"`
Schedule string `json:"schedule"`
Category string `json:"category"` // monitor / cleanup
Runnable bool `json:"runnable"`
Enabled bool `json:"enabled"`
}
// Item API 列表项(元数据 + 运行状态) // Logan.Li
type Item struct {
Def
LastRunAt *time.Time `json:"last_run_at,omitempty"`
LastStatus string `json:"last_status,omitempty"`
LastMessage string `json:"last_message,omitempty"`
RunCount int64 `json:"run_count"`
LastDurationMs int `json:"last_duration_ms"`
Extra any `json:"extra,omitempty"`
}
var (
mu sync.RWMutex
db *gorm.DB
catalog = []Def{
{
Code: "alert_engine", Name: "告警引擎扫描",
Description: "扫描 Agent 指标阈值、主机离线、端口策略与日志告警规则,触发通知与自动恢复",
Schedule: "每 30 秒", Category: "monitor", Runnable: false, Enabled: true,
},
{
Code: "middleware_monitor", Name: "中间件监控采集",
Description: "直连 Redis/MySQL 等中间件目标采集指标并写入监控库",
Schedule: "每 30 秒", Category: "monitor", Runnable: false, Enabled: true,
},
{
Code: "agent_report_cleanup", Name: "Agent 上报清理",
Description: "清理 7 天前的 Agent 指标上报原始数据,防止表膨胀",
Schedule: "每 24 小时", Category: "cleanup", Runnable: true, Enabled: true,
},
{
Code: "audit_log_cleanup", Name: "审计日志清理",
Description: "按系统设置中的保留天数清理操作审计、登录日志、终端会话与命令记录",
Schedule: "每 24 小时", Category: "cleanup", Runnable: true, Enabled: true,
},
{
Code: "snmp_metrics_cleanup", Name: "SNMP 指标清理",
Description: "清理 7 天前的 SNMP 采集指标历史,防止表膨胀",
Schedule: "每 24 小时", Category: "cleanup", Runnable: true, Enabled: true,
},
}
)
// Init 注入数据库连接,并确保状态行存在 // Logan.Li
func Init(gdb *gorm.DB) {
mu.Lock()
db = gdb
mu.Unlock()
ensureStatusRows(gdb)
}
func ensureStatusRows(gdb *gorm.DB) {
for _, d := range catalog {
var row model.SystemTaskStatus
if gdb.Where("code = ?", d.Code).First(&row).Error == gorm.ErrRecordNotFound {
gdb.Create(&model.SystemTaskStatus{Code: d.Code, LastStatus: "pending"})
}
}
}
// Catalog 返回任务定义列表 // Logan.Li
func Catalog() []Def {
out := make([]Def, len(catalog))
copy(out, catalog)
return out
}
// Record 记录任务执行结果 // Logan.Li
func Record(code, status, message string, duration time.Duration) {
mu.RLock()
gdb := db
mu.RUnlock()
if gdb == nil {
return
}
now := time.Now()
updates := map[string]interface{}{
"last_run_at": now,
"last_status": status,
"last_message": truncate(message, 480),
"last_duration_ms": int(duration.Milliseconds()),
"run_count": gorm.Expr("run_count + 1"),
}
gdb.Model(&model.SystemTaskStatus{}).Where("code = ?", code).Updates(updates)
}
func truncate(s string, max int) string {
if len(s) <= max {
return s
}
return s[:max]
}
// List 合并元数据与库中状态 // Logan.Li
func List(gdb *gorm.DB, extras map[string]any) []Item {
ensureStatusRows(gdb)
var rows []model.SystemTaskStatus
gdb.Find(&rows)
statusMap := make(map[string]model.SystemTaskStatus, len(rows))
for _, r := range rows {
statusMap[r.Code] = r
}
items := make([]Item, 0, len(catalog))
for _, d := range catalog {
item := Item{Def: d}
if st, ok := statusMap[d.Code]; ok {
item.LastRunAt = st.LastRunAt
item.LastStatus = st.LastStatus
item.LastMessage = st.LastMessage
item.RunCount = st.RunCount
item.LastDurationMs = st.LastDurationMs
}
if extras != nil {
if ex, ok := extras[d.Code]; ok {
item.Extra = ex
}
}
// 审计清理任务补充保留天数 // Logan.Li
if d.Code == "audit_log_cleanup" {
days := auditcleaner.GetRetentionDays(gdb)
if item.Extra == nil {
item.Extra = map[string]any{"retention_days": days}
}
}
if d.Code == "snmp_metrics_cleanup" && item.Extra == nil {
item.Extra = map[string]any{"retention_days": snmppkg.MetricsRetentionDays}
}
items = append(items, item)
}
return items
}
// RunManual 手动触发可运行的系统任务 // Logan.Li
func RunManual(gdb *gorm.DB, driver, code string) (Item, error) {
def, ok := findDef(code)
if !ok {
return Item{}, fmt.Errorf("未知系统任务: %s", code)
}
if !def.Runnable {
return Item{}, fmt.Errorf("任务 %s 不支持手动执行", def.Name)
}
start := time.Now()
var msg string
var err error
switch code {
case "agent_report_cleanup":
sql := database.CleanupAgentReportsSQL(driver)
r := gdb.Exec(sql)
err = r.Error
msg = fmt.Sprintf("清理 %d 条 Agent 上报记录", r.RowsAffected)
case "audit_log_cleanup":
var result auditcleaner.Result
result, err = auditcleaner.RunFromSettings(gdb)
if err == nil {
msg = fmt.Sprintf("清理 %d 条超期记录(保留 %d 天)", result.Total, result.RetentionDays)
now := time.Now()
gdb.Model(&model.AuditSettings{}).Where("id > 0").Updates(map[string]interface{}{
"last_cleanup_at": now,
"last_cleanup_summary": result.SummaryJSON(),
})
}
case "snmp_metrics_cleanup":
sql := database.CleanupSNMPMetricsSQL(driver, snmppkg.MetricsRetentionDays)
r := gdb.Exec(sql)
err = r.Error
msg = fmt.Sprintf("清理 %d 条 SNMP 指标(保留 %d 天)", r.RowsAffected, snmppkg.MetricsRetentionDays)
default:
return Item{}, fmt.Errorf("任务 %s 未实现手动执行", code)
}
duration := time.Since(start)
status := "success"
if err != nil {
status = "failed"
msg = err.Error()
}
Record(code, status, msg, duration)
items := List(gdb, nil)
for _, it := range items {
if it.Code == code {
return it, err
}
}
return Item{Def: def}, err
}
func findDef(code string) (Def, bool) {
for _, d := range catalog {
if d.Code == code {
return d, true
}
}
return Def{}, false
}