feat(audit): F1+F2 告警静默规则标记完成 + 审计日志CSV导出实现

F1 告警静默规则 [done]:
- 后端 alert_silence.go: CRUD + Toggle (onetime/daily/weekly 三模式)
- 告警引擎 isSilenced 集成 (alert_engine.go)
- 前端 SilenceRuleTab + SilenceDialog
- 路由 /alerts/silences 已注册

F2 审计日志导出 [done]:
- 重写 ExportAuditLogs: 真正的 CSV 导出 (encoding/csv + UTF-8 BOM)
- 支持 module/action/username/startTime/endTime 筛选
- 导出限制 10000 条防溢出,自动记录审计日志
- JSON 格式向后兼容 (format=json)
- 前端 exportLogs 传递查询参数

同时包含累积的前期工作:
- 目录重构 internal/ -> controller/model/router (v0.55-v0.57)
- 后端分层架构 dto/repository/service/utils
- 审计模块全功能 (终端审计/操作审计/登录日志)
- asciicast v2 录制格式
- SFTP 文件浏览器
- StatStrip 紧凑统计条
- 在线会话自动刷新
- LiveMonitor WebSocket 诊断修复

验证: go build ✅ | npm run build ✅ | Docker 部署 ✅ | 10 回归端点 ✅ | CSV 导出 ✅ | 静默 CRUD ✅
This commit is contained in:
attacker committed 2026-06-25 20:16:30 +08:00
1 parent e57718998a
commit 77c234b060
314 files changed
+10050 -4065

No files matched your search

+624
View File
@@ -0,0 +1,624 @@
package controller
import (
"encoding/json"
"fmt"
"math"
"net/http"
"sort"
"strconv"
"strings"
"time"
"ezops/model"
ezredis "ezops/pkg/redis"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// AgentHandler 处理 Agent 上报数据 // Logan.Li
type AgentHandler struct {
db *gorm.DB
cache ezredis.Cache // 可选 Redis 缓存(用于热数据存储)
hubMetrics *MetricsHub // WebSocket 广播 metrics 更新
}
func NewAgentHandler(db *gorm.DB, cache ezredis.Cache, hub *MetricsHub) *AgentHandler {
return &AgentHandler{db: db, cache: cache, hubMetrics: hub}
}
// ReceiveReport 接收 Agent 上报的指标数据
// POST /api/v1/agent/report
func (h *AgentHandler) ReceiveReport(c *gin.Context) {
var req model.AgentReportRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
return
}
// 解析上报时间
reportTime := time.Now()
if req.Timestamp != "" {
if t, err := time.Parse(time.RFC3339, req.Timestamp); err == nil {
reportTime = t
}
}
// 序列化磁盘和端口信息
diskJSON, _ := json.Marshal(req.Disks)
portJSON, _ := json.Marshal(req.Ports)
netJSON, _ := json.Marshal(req.Network)
dioJSON, _ := json.Marshal(req.DiskIO)
// v2.0 新增
swapJSON, _ := json.Marshal(req.Swap)
processTopJSON, _ := json.Marshal(req.ProcessTop)
fdJSON, _ := json.Marshal(req.FD)
csJSON, _ := json.Marshal(req.ContextSwitches)
heartbeatJSON, _ := json.Marshal(req.Heartbeat)
// 存储到数据库
report := model.AgentReport{
Hostname: req.Hostname,
CPUUsage: req.CPU.UsagePercent,
CPUIOwait: req.CPU.IOWait,
CPUCores: req.CPU.CoreCount,
Load1: req.CPU.Load1,
Load5: req.CPU.Load5,
Load15: req.CPU.Load15,
MemTotal: req.Memory.TotalGB,
MemUsed: req.Memory.UsedGB,
MemUsage: req.Memory.UsagePercent,
MemAvailable: req.Memory.AvailableGB,
MemSwapUsed: req.Memory.SwapUsedGB,
DiskInfo: string(diskJSON),
PortInfo: string(portJSON),
NetInfo: string(netJSON),
DiskIOInfo: string(dioJSON),
// v2.0 新增
SwapInfo: string(swapJSON),
ProcessTopInfo: string(processTopJSON),
FDInfo: string(fdJSON),
CSInfo: string(csJSON),
HeartbeatInfo: string(heartbeatJSON),
ReportTime: reportTime,
}
if err := h.db.Create(&report).Error; err != nil {
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "保存失败: " + err.Error()})
return
}
// 同步更新 servers 表的实时指标(三级匹配:hostname → client IP → hostname LIKE)
diskAvg := avgDiskUsage(req.Disks)
// 计算硬件规格(四舍五入到整数,避免 31.34→31GB 的显示偏差)
totalDiskGB := 0.0
for _, d := range req.Disks {
totalDiskGB += d.TotalGB
}
memGB := math.Ceil(req.Memory.TotalGB) // 31.34 → 32, 62.79 → 63(物理内存向上取整)
memMB := int64(memGB * 1024) // GB → MB
diskGB := int64(totalDiskGB + 0.5) // 39.99 → 40
clientIP := c.ClientIP()
// 三级匹配:优先用 Agent 上报的 IP,其次用请求源 IP,最后用报文的 hostname
matchIP := req.IP
if matchIP == "" {
matchIP = clientIP
}
// 直接 SQL 更新 asset_servers 监控字段
// 若 AgentVersion 非空则同步写入,否则保留数据库中已有版本号(防止旧版Agent空字段覆盖已设置的版本)
updateSQL := `UPDATE asset_servers SET
cpu_usage = ?, cpu_iowait = ?, memory_usage = ?, disk_usage = ?,
cpu = ?, memory = ?, disk = ?,
status = ?, agent_status = ?` + func() string {
if req.AgentVersion != "" {
return `, agent_version = ?`
}
return ``
}() + `,
hostname = CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END,
updated_at = NOW()`
args := []interface{}{
req.CPU.UsagePercent, req.CPU.IOWait, req.Memory.UsagePercent, diskAvg,
req.CPU.CoreCount, memMB, diskGB,
"running", "online",
}
if req.AgentVersion != "" {
args = append(args, req.AgentVersion)
}
args = append(args,
// CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END
req.Hostname, matchIP, req.Hostname,
)
if req.System != nil {
if req.System.OSVersion != "" {
updateSQL += ", os = ?"
args = append(args, req.System.OSVersion)
}
if req.System.Kernel != "" {
updateSQL += ", kernel = ?"
args = append(args, req.System.Kernel)
}
if req.System.Uptime > 0 {
updateSQL += ", uptime = ?"
args = append(args, req.System.Uptime)
}
}
updateSQL += " WHERE hostname = ? OR ip = ?"
args = append(args, req.Hostname, matchIP)
result := h.db.Exec(updateSQL, args...)
// 如果 hostname 和 IP 都没匹配到,尝试用 hostname 反向匹配 IP 字段
if result.RowsAffected == 0 && req.Hostname != matchIP {
updateSQL2 := updateSQL + " OR ip = ?"
args2 := append(args, req.Hostname)
result = h.db.Exec(updateSQL2, args2...)
}
// 如果还是没匹配到 —— 自动注册新服务器(UPSERT 防重复)
if result.RowsAffected == 0 {
osName := ""
if req.System != nil {
osName = req.System.OSVersion
}
newServer := model.Server{
Hostname: req.Hostname,
IP: matchIP,
Port: 22,
Username: "root",
AuthType: "password",
OS: osName,
CPU: req.CPU.CoreCount,
Memory: memMB,
Disk: diskGB,
Status: "running",
AgentStatus: "online",
AgentVersion: req.AgentVersion,
CPUUsage: req.CPU.UsagePercent,
MemoryUsage: req.Memory.UsagePercent,
DiskUsage: diskAvg,
}
// ON DUPLICATE KEY UPDATE — 若 hostname 已存在则更新字段(防重复注册)
if err := h.db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "hostname"}},
DoUpdates: clause.AssignmentColumns([]string{
"ip", "os", "cpu", "memory", "disk",
"status", "agent_status",
"cpu_usage", "memory_usage", "disk_usage",
"updated_at",
}),
}).Create(&newServer).Error; err != nil {
fmt.Printf("[agent] 自动注册主机 %q (IP: %s) 失败: %v\n", req.Hostname, matchIP, err)
} else {
fmt.Printf("[agent] 自动注册新主机: %q (IP: %s, CPU=%d核, 内存=%dMB, 磁盘=%dGB, OS=%q)\n",
req.Hostname, matchIP, req.CPU.CoreCount, memMB, diskGB, osName)
}
}
// WebSocket 广播:向所有前端客户端推送该主机的最新指标 // Logan.Li
if h.hubMetrics != nil && result.RowsAffected > 0 {
h.hubMetrics.Broadcast(MetricsUpdate{
Type: "metrics_update",
Hostname: req.Hostname,
CPU: req.CPU.UsagePercent,
Memory: req.Memory.UsagePercent,
Disk: diskAvg,
Status: "running",
TCPEstablished: req.Network.TCPEstablished,
TCPTimeWait: req.Network.TCPTimeWait,
NetRxBPS: req.Network.NetRxBPS,
NetTxBPS: req.Network.NetTxBPS,
DiskReadBPS: req.DiskIO.ReadBPS,
DiskWriteBPS: req.DiskIO.WriteBPS,
})
}
// 写入 Redis 有序集合(热数据,保留7天)
if h.cache != nil {
timestamp := float64(reportTime.Unix())
ttl := 7 * 24 * time.Hour // 7天
// 使用 reportTime.UnixNano() 确保 member 唯一(避免相同值碰撞)
h.writeHotMetric(c, "metrics:cpu:"+req.Hostname, req.CPU.UsagePercent, timestamp, ttl, reportTime)
h.writeHotMetric(c, "metrics:mem:"+req.Hostname, req.Memory.UsagePercent, timestamp, ttl, reportTime)
h.writeHotMetric(c, "metrics:load1:"+req.Hostname, req.CPU.Load1, timestamp, ttl, reportTime)
}
// 保存成功 — 响应中携带探针配置给 Agent
resp := h.makeProbeConfigResponse(req.Hostname, 200, "ok", req.ProbeResults)
c.JSON(http.StatusOK, resp)
}
// writeHotMetric 将单个指标写入 Redis 有序集合并维护 TTL 和过期数据清理
func (h *AgentHandler) writeHotMetric(c *gin.Context, key string, value float64, score float64, ttl time.Duration, reportTime time.Time) {
ctx := c.Request.Context()
// member 格式: "value:nano_timestamp" 保证唯一性
member := fmt.Sprintf("%.4f:%d", value, reportTime.UnixNano())
if err := h.cache.ZAdd(ctx, key, ezredis.SortedSetMember{
Score: score,
Member: member,
}); err != nil {
// 静默失败,不影响主流程
return
}
// 设置 key 过期时间
_ = h.cache.Expire(ctx, key, ttl)
// 清理 7 天前的旧数据
cutoff := float64(time.Now().Add(-ttl).Unix())
_ = h.cache.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%.0f", cutoff))
}
// GetLatestReports 获取所有服务器的 Agent 最新状态(含未安装 Agent 的主机)
// GET /api/v1/agent/latest
func (h *AgentHandler) GetLatestReports(c *gin.Context) {
var results []model.AgentLatestResponse
threeMinAgo := time.Now().Add(-3 * time.Minute)
// 以 servers 表为主,LEFT JOIN 最近 3 分钟内的 Agent 上报
// 未装 Agent 的主机也会显示(agent_status='not_installed'),打破"不显示→无法安装"的循环
h.db.Raw(`
SELECT
s.id as server_id,
s.hostname,
COALESCE(s.ip, '') as ip,
COALESCE(a.cpu_usage, 0) as cpu_usage,
COALESCE(a.cpu_cores, 0) as cpu_cores,
COALESCE(a.mem_usage, 0) as mem_usage,
COALESCE(a.mem_total, 0) as mem_total,
COALESCE(a.mem_used, 0) as mem_used,
COALESCE(s.disk_usage, 0) as disk_usage,
a.disk_info,
a.port_info,
a.net_info,
a.disk_io_info,
a.swap_info,
a.process_top_info,
a.fd_info,
a.cs_info,
a.heartbeat_info,
a.report_time,
COALESCE(s.agent_version, '') as agent_version,
CASE
WHEN a.hostname IS NOT NULL THEN COALESCE(s.agent_status, 'online')
ELSE 'not_installed'
END as agent_status,
COALESCE(s.os, '') as os,
COALESCE(s.kernel, '') as kernel,
COALESCE(s.uptime, 0) as uptime,
COALESCE(s.disk, 0) as disk_total,
COALESCE(r.name, '') as region,
CASE WHEN s.password != '' OR s.ssh_key != '' THEN true ELSE false END as has_auth
FROM asset_servers s
LEFT JOIN (
SELECT * FROM monitor_agent_reports
WHERE id IN (
SELECT MAX(id) FROM monitor_agent_reports
WHERE report_time >= ?
GROUP BY hostname
)
) a ON a.hostname = s.hostname
LEFT JOIN asset_regions r ON r.id = s.region_id
ORDER BY a.report_time DESC, s.id ASC
`, threeMinAgo).Scan(&results)
if results == nil {
results = []model.AgentLatestResponse{}
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "success",
Data: results,
})
}
// GetHostHistory 获取指定主机的历史指标
// GET /api/v1/agent/history/:hostname?hours=1
func (h *AgentHandler) GetHostHistory(c *gin.Context) {
hostname := c.Param("hostname")
hours := 1
if v := c.Query("hours"); v != "" {
fmt.Sscanf(v, "%d", &hours)
}
since := time.Now().Add(-time.Duration(hours) * time.Hour)
var reports []model.AgentReport
h.db.Where("hostname = ? AND report_time >= ?", hostname, since).
Order("report_time ASC").
Find(&reports)
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "success",
Data: reports,
})
}
// AggregateMetrics 聚合 Redis 热数据到 MySQL klines
// POST /api/v1/agent/aggregate
func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
var req model.AggregateReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
return
}
if h.cache == nil {
c.JSON(http.StatusServiceUnavailable, model.Response{Code: 503, Message: "Redis 缓存不可用"})
return
}
// 支持的聚合间隔及对应的回溯时长
intervals := []struct {
name string
duration time.Duration
}{
{"1m", 1 * time.Minute},
{"5m", 5 * time.Minute},
{"3m", 3 * time.Minute},
{"1h", 1 * time.Hour},
{"6h", 6 * time.Hour},
{"1d", 24 * time.Hour},
}
redisKey := fmt.Sprintf("metrics:%s:%s", req.Metric, req.Hostname)
ctx := c.Request.Context()
var totalUpserted int64
for _, iv := range intervals {
// 回溯窗口:取 interval × 200 倍,确保覆盖足够多的数据点
lookback := iv.duration * 200
now := time.Now()
since := now.Add(-lookback)
minScore := fmt.Sprintf("%d", since.Unix())
maxScore := fmt.Sprintf("%d", now.Unix()+1)
members, err := h.cache.ZRangeByScore(ctx, redisKey, minScore, maxScore)
if err != nil || len(members) == 0 {
continue
}
// 按间隔分桶
buckets := make(map[int64][]float64) // bucketTime (unix) → values
for _, m := range members {
ts := int64(m.Score)
bucketStart := (ts / int64(iv.duration.Seconds())) * int64(iv.duration.Seconds())
val := parseMemberValue(m.Member)
buckets[bucketStart] = append(buckets[bucketStart], val)
}
// 计算 max/min/avg/p95
for bucketTs, values := range buckets {
if len(values) == 0 {
continue
}
maxV := values[0]
minV := values[0]
sum := 0.0
for _, v := range values {
sum += v
if v > maxV {
maxV = v
}
if v < minV {
minV = v
}
}
avg := sum / float64(len(values))
// 计算 P95(排序后取 95% 位置的值)
p95 := maxV
if len(values) >= 20 {
sorted := make([]float64, len(values))
copy(sorted, values)
sort.Float64s(sorted)
idx := int(float64(len(sorted)) * 0.95)
if idx >= len(sorted) {
idx = len(sorted) - 1
}
p95 = sorted[idx]
}
agg := model.MetricsAggregate{
Hostname: req.Hostname,
Metric: req.Metric,
Interval: iv.name,
Avg: math.Round(avg*100) / 100,
MaxV: math.Round(maxV*100) / 100,
MinV: math.Round(minV*100) / 100,
P95: math.Round(p95*100) / 100,
Count: len(values),
BucketTime: time.Unix(bucketTs, 0),
}
// Upsert: hostname + metric + interval + bucket_time 唯一
result := h.db.Clauses(clause.OnConflict{
Columns: []clause.Column{
{Name: "hostname"},
{Name: "metric"},
{Name: "interval"},
{Name: "bucket_time"},
},
DoUpdates: clause.AssignmentColumns([]string{
"avg", "max_v", "min_v", "p95", "count",
}),
}).Create(&agg)
if result.Error == nil {
totalUpserted++
}
}
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: fmt.Sprintf("聚合完成,写入/更新 %d 条记录", totalUpserted),
Data: gin.H{"upserted": totalUpserted},
})
}
// GetAggregate 查询聚合数据(原 GetKline)
// GET /api/v1/agent/aggregate?hostname=&metric=&interval=&from=&to=
func (h *AgentHandler) GetAggregate(c *gin.Context) {
var req model.AggregateQueryReq
if err := c.ShouldBindQuery(&req); err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
from, err := time.Parse(time.RFC3339, req.From)
if err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "from 格式错误,需 RFC3339: " + err.Error()})
return
}
to, err := time.Parse(time.RFC3339, req.To)
if err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "to 格式错误,需 RFC3339: " + err.Error()})
return
}
var items []model.MetricsAggregate
result := h.db.Where("hostname = ? AND metric = ? AND `interval` = ? AND bucket_time >= ? AND bucket_time <= ?",
req.Hostname, req.Metric, req.Interval, from, to).
Order("bucket_time ASC").
Find(&items)
if result.Error != nil {
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "查询失败: " + result.Error.Error()})
return
}
if items == nil {
items = []model.MetricsAggregate{}
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "success",
Data: items,
})
}
// parseMemberValue 从 member 字符串 "45.2300:1234567890123456789" 中提取浮点值
func parseMemberValue(member string) float64 {
idx := strings.Index(member, ":")
if idx < 0 {
v, _ := strconv.ParseFloat(member, 64)
return v
}
v, _ := strconv.ParseFloat(member[:idx], 64)
return v
}
// avgDiskUsage 计算磁盘平均使用率
func avgDiskUsage(disks []model.DiskMetric) float64 {
if len(disks) == 0 {
return 0
}
var sum float64
for _, d := range disks {
sum += d.UsagePercent
}
return math.Round(sum/float64(len(disks))*100) / 100
}
// ── 端口监控策略 ── // Logan.Li
// GetPortPolicyForAgent Agent 拉取本机端口策略(无需鉴权,通过 hostname 查询)
// GET /api/v1/agent/port-policy?hostname=xxx
func (h *AgentHandler) GetPortPolicyForAgent(c *gin.Context) {
hostname := c.Query("hostname")
if hostname == "" {
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
PortFilterMode: "whitelist",
PortWhitelist: []int{},
}})
return
}
var policy model.AgentPortPolicy
if err := h.db.Where("hostname = ?", hostname).First(&policy).Error; err != nil {
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
Hostname: hostname,
PortFilterMode: "whitelist",
PortWhitelist: []int{},
}})
return
}
var whitelist []int
json.Unmarshal([]byte(policy.PortWhitelist), &whitelist)
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
Hostname: policy.Hostname,
PortFilterMode: policy.PortFilterMode,
PortWhitelist: whitelist,
}})
}
// GetPortPolicy 前端获取主机的端口监控策略
// GET /api/v1/agent/port-policies/:hostname
func (h *AgentHandler) GetPortPolicy(c *gin.Context) {
hostname := c.Param("hostname")
var policy model.AgentPortPolicy
var whitelist []int
if err := h.db.Where("hostname = ?", hostname).First(&policy).Error; err != nil {
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
Hostname: hostname,
PortFilterMode: "whitelist",
PortWhitelist: []int{},
}})
return
}
json.Unmarshal([]byte(policy.PortWhitelist), &whitelist)
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
Hostname: policy.Hostname,
PortFilterMode: policy.PortFilterMode,
PortWhitelist: whitelist,
}})
}
// SavePortPolicy 更新/创建主机的端口监控策略
// PUT /api/v1/agent/port-policies/:hostname
func (h *AgentHandler) SavePortPolicy(c *gin.Context) {
hostname := c.Param("hostname")
var dto model.AgentPortPolicyDTO
if err := c.ShouldBindJSON(&dto); err != nil {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
return
}
whitelistJSON := "[]"
if len(dto.PortWhitelist) > 0 {
b, _ := json.Marshal(dto.PortWhitelist)
whitelistJSON = string(b)
}
policy := model.AgentPortPolicy{
Hostname: hostname,
PortFilterMode: dto.PortFilterMode,
PortWhitelist: whitelistJSON,
}
if err := h.db.Where(model.AgentPortPolicy{Hostname: hostname}).
Assign(policy).
FirstOrCreate(&policy).Error; err != nil {
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "保存失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "端口策略已保存"})
}