perf: 修复 Agent 校验 CPU 热点并实现指标分层聚合与频率下发
CPU 根因:每个 Agent 请求都做一次 bcrypt 校验(实测 72ms), 30 台主机 18 req/s × 72ms ≈ 1.35 核,实测容器 CPU 135%。 - agentauth: 校验结果按 Token 哈希缓存(TTL 5min,轮换即失效), last_used_at 写入按 60s 节流。实测 CPU 135% → 1.6%,请求延迟 103ms → 1.6ms - metricagg: 新增后台逐级降采样聚合 raw→1m/5m→1h→6h/1d,取代前端手动触发 - metrics_aggregate: 补 uniq_metric_agg 唯一索引(原 idx_agg 非唯一导致 upsert 插重复行) - agent 合并配置接口 GET /agent/config:一次返回任务+探针+端口策略+频率, 配置轮询 5s→30s;服务端可下发 report_interval,Agent 热更新 ticker 无需重装 - retention: agent_reports 保留期 7d→1d(趋势由聚合表承担) - redis: 增加 maxmemory 512mb + volatile-lru - fix(upgrade): 升级脚本改为 fail-fast,修复下载失败仍报 UPGRADE_OK 的静默谎报 // Logan.Li
This commit is contained in:
1 parent
93abf81e40
commit
59320aa410
23 files changed
+902
-29
No files matched your search
+1
-1
@@ -164,4 +164,4 @@ cd agent
|
||||
make # CGO_ENABLED=0 静态编译当前平台
|
||||
```
|
||||
|
||||
顶层 `./run.sh build agent` 会交叉编译 `linux-amd64`、`linux-arm64`、`darwin-arm64`、`windows-amd64.exe`。Docker 镜像会携带当前目标架构 Linux Agent 和 Windows AMD64 Agent。// Logan.Li
|
||||
顶层 `./run.sh build agent` 会交叉编译 `linux-amd64`、`linux-arm64`、`windows-amd64.exe`。Docker 镜像内置同样的三个平台 Agent 二进制,覆盖主流服务器架构。// Logan.Li
|
||||
@@ -106,7 +106,7 @@ curl -fSL -o /tmp/aiops-agent "${BINARY_URL}" || {
|
||||
echo "❌ 下载失败: ${BINARY_URL}"
|
||||
echo " 请检查: 1) 后端地址是否正确 2) 该平台是否有对应二进制"
|
||||
if [ "$OS_LOWER" = "darwin" ]; then
|
||||
echo " 提示: macOS 二进制需要后端编译 darwin 版本 (make build-darwin-arm64)"
|
||||
echo " 提示: 平台不再提供 macOS Agent 二进制,服务器采集请使用 Linux/Windows 主机"
|
||||
fi
|
||||
exit 1
|
||||
}
|
||||
|
||||
+72
-13
@@ -35,7 +35,7 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
version = "v2.1.0"
|
||||
version = "v2.2.0"
|
||||
buildTime = "unknown"
|
||||
)
|
||||
|
||||
@@ -292,33 +292,92 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
const pollInterval = 5 * time.Second
|
||||
// ── 配置轮询 + 上报循环(v2.2)────────────────────────────────
|
||||
//
|
||||
// 旧版每 5 秒分别拉取 tasks/poll、health-probes/config、port-policy 三个接口,
|
||||
// 配置类请求量是上报本身的 6 倍;现改为单次 GET /agent/config,
|
||||
// 并由服务端下发 report_interval,Agent 热更新 ticker、无需重装。 // Logan.Li
|
||||
const minInterval = 5 * time.Second
|
||||
const fallbackConfigPollInterval = 30 * time.Second
|
||||
intervalUpdates := make(chan time.Duration, 1)
|
||||
|
||||
configTicker := time.NewTicker(fallbackConfigPollInterval)
|
||||
go func() {
|
||||
pollAndExecute()
|
||||
pullProbes()
|
||||
pullPortPolicy()
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
defer configTicker.Stop()
|
||||
runConfig := func() {
|
||||
merged, err := taskClient.PollConfig(ctx)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
// 服务端尚未升级(或暂时失败)时回退旧的分接口模式,保证兼容。 // Logan.Li
|
||||
pollAndExecute()
|
||||
pullProbes()
|
||||
pullPortPolicy()
|
||||
return
|
||||
}
|
||||
|
||||
// 执行服务端下发的任务。 // Logan.Li
|
||||
for _, t := range merged.Tasks {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
result := taskExecutor.ExecuteTask(ctx, t)
|
||||
if err := taskClient.Submit(ctx, result); err != nil {
|
||||
logger.Printf("任务 %s 结果回传失败: %v", t.TaskID, err)
|
||||
}
|
||||
}
|
||||
if len(merged.HealthProbes) > 0 {
|
||||
probeConfigs = merged.HealthProbes
|
||||
}
|
||||
if merged.PortPolicy != nil {
|
||||
portCollector.SetPolicy(merged.PortPolicy)
|
||||
}
|
||||
|
||||
// 服务端下发的上报间隔 → 通知主循环热更新 ticker。 // Logan.Li
|
||||
if merged.ReportInterval > 0 {
|
||||
next := time.Duration(merged.ReportInterval) * time.Second
|
||||
if next < minInterval {
|
||||
next = minInterval
|
||||
}
|
||||
select {
|
||||
case intervalUpdates <- next:
|
||||
default:
|
||||
}
|
||||
}
|
||||
// 服务端下发的配置轮询间隔 → 调整本循环节奏。 // Logan.Li
|
||||
if merged.ConfigPollInterval > 0 {
|
||||
next := time.Duration(merged.ConfigPollInterval) * time.Second
|
||||
if next >= minInterval && next != fallbackConfigPollInterval {
|
||||
configTicker.Reset(next)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
runConfig()
|
||||
for {
|
||||
select {
|
||||
case <-configTicker.C:
|
||||
runConfig()
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// 上报循环:间隔可被服务端热更新。 // Logan.Li
|
||||
time.Sleep(startupJitter)
|
||||
collectAndReport()
|
||||
ticker := time.NewTicker(time.Duration(cfg.Interval) * time.Second)
|
||||
defer ticker.Stop()
|
||||
reportTicker := time.NewTicker(time.Duration(cfg.Interval) * time.Second)
|
||||
defer reportTicker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-reportTicker.C:
|
||||
collectAndReport()
|
||||
case next := <-intervalUpdates:
|
||||
// 重新抖动,避免多台主机在服务端下发同值后被对齐成同一时刻上报。 // Logan.Li
|
||||
reportTicker.Reset(next)
|
||||
logger.Printf("服务端下发新的上报间隔: %s", next)
|
||||
case <-ctx.Done():
|
||||
logger.Println("收到退出信号,Agent 正常关闭")
|
||||
return
|
||||
|
||||
@@ -20,6 +20,8 @@ const (
|
||||
resultPathFmt = "/api/v1/agent/tasks/%s/result"
|
||||
probePath = "/api/v1/agent/health-probes/config"
|
||||
portPolicyPath = "/api/v1/agent/port-policy"
|
||||
// configPath 合并配置接口(v2.2):一次拉取任务+探针+端口策略+采集频率。 // Logan.Li
|
||||
configPath = "/api/v1/agent/config"
|
||||
)
|
||||
|
||||
// Client 负责与后端任务/探针接口通信。
|
||||
@@ -90,3 +92,32 @@ func (c *Client) PollPortPolicy(ctx context.Context) (*model.PortPolicy, error)
|
||||
}
|
||||
return &wrapper.Data, nil
|
||||
}
|
||||
|
||||
// AgentConfig 合并配置响应(v2.2)。
|
||||
//
|
||||
// 旧版 Agent 每 5 秒要分别请求 tasks/poll、health-probes/config、port-policy
|
||||
// 三个接口,配置类请求量是上报本身的 6 倍;合并后单次请求即可拿到全部配置。 // Logan.Li
|
||||
type AgentConfig struct {
|
||||
ConfigVersion string `json:"config_version"`
|
||||
ReportInterval int `json:"report_interval"`
|
||||
ConfigPollInterval int `json:"config_poll_interval"`
|
||||
Tasks []Task `json:"tasks"`
|
||||
HealthProbes []model.ProbeConfig `json:"health_probes"`
|
||||
PortPolicy *model.PortPolicy `json:"port_policy,omitempty"`
|
||||
}
|
||||
|
||||
// PollConfig 一次性拉取任务 + 探针 + 端口策略 + 采集频率(v2.2 合并接口)。
|
||||
// 若服务端尚未升级(404),调用方应回退到分接口拉取。 // Logan.Li
|
||||
func (c *Client) PollConfig(ctx context.Context) (*AgentConfig, error) {
|
||||
path := configPath + "?hostname=" + url.QueryEscape(c.hostname)
|
||||
|
||||
var wrapper struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Data AgentConfig `json:"data"`
|
||||
}
|
||||
if err := c.client.DoJSON(ctx, "GET", path, nil, &wrapper); err != nil {
|
||||
return nil, fmt.Errorf("配置拉取失败: %w", err)
|
||||
}
|
||||
return &wrapper.Data, nil
|
||||
}
|
||||
+5
-3
@@ -67,12 +67,12 @@ RUN --mount=type=cache,target=/go/pkg/mod \
|
||||
# 复制 agent 源码
|
||||
COPY agent/ ./
|
||||
|
||||
# 交叉编译三个平台的 agent 二进制(注入版本号)
|
||||
# 交叉编译主流服务器平台 agent 二进制(Linux amd64 / Linux arm64 / Windows amd64,注入版本号)
|
||||
RUN --mount=type=cache,target=/go/pkg/mod \
|
||||
--mount=type=cache,target=/root/.cache/go-build \
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags="-s -w -X main.version=${VERSION} -X main.buildTime=${BUILD_TIME}" -o bin/aiops-agent-linux-amd64 . && \
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags="-s -w -X main.version=${VERSION} -X main.buildTime=${BUILD_TIME}" -o bin/aiops-agent-linux-arm64 . && \
|
||||
CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -ldflags="-s -w -X main.version=${VERSION} -X main.buildTime=${BUILD_TIME}" -o bin/aiops-agent-darwin-arm64 .
|
||||
CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags="-s -w -X main.version=${VERSION} -X main.buildTime=${BUILD_TIME}" -o bin/aiops-agent-windows-amd64.exe .
|
||||
|
||||
# ==================== Stage 3: Runtime ====================
|
||||
# 固定版本避免 :latest 每次变动导致全量 cache 失效 // Logan.Li
|
||||
@@ -106,8 +106,10 @@ COPY --from=builder /app/aiops .
|
||||
# 创建运行时目录(agent 二进制 + 安装脚本)
|
||||
RUN mkdir -p /app/agent/bin /app/agent/deploy
|
||||
|
||||
# 复制 agent 二进制(U4: 仅 linux-amd64 内置, arm64/darwin 通过下载 API 提供)// Logan.Li
|
||||
# 复制 agent 二进制(Linux amd64 内置,其余平台通过下载 API 提供)// Logan.Li
|
||||
COPY --from=agent-builder /agent/bin/aiops-agent-linux-amd64 /app/agent/bin/
|
||||
COPY --from=agent-builder /agent/bin/aiops-agent-linux-arm64 /app/agent/bin/
|
||||
COPY --from=agent-builder /agent/bin/aiops-agent-windows-amd64.exe /app/agent/bin/
|
||||
|
||||
# 复制 agent 安装脚本和 systemd service 模板
|
||||
COPY agent/deploy/install.sh /app/agent/deploy/
|
||||
|
||||
@@ -97,7 +97,9 @@ func Load() *Config {
|
||||
viper.SetDefault("database.max_idle", 10)
|
||||
viper.SetDefault("jwt.expire_hour", 24)
|
||||
viper.SetDefault("agent.enforce_auth", false)
|
||||
viper.SetDefault("retention.agent_reports_days", 7)
|
||||
// 原始 Agent 上报仅作短期排障用;长期趋势由 monitor_metric_aggregates 承担,
|
||||
// 故原始表保留期收敛到 1 天,避免高频表无限膨胀。 // Logan.Li
|
||||
viper.SetDefault("retention.agent_reports_days", 1)
|
||||
viper.SetDefault("retention.metric_history_days", 30)
|
||||
viper.SetDefault("retention.metric_aggregate_days", 365)
|
||||
viper.SetDefault("retention.snmp_samples_days", 30)
|
||||
|
||||
@@ -0,0 +1,198 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"ezops/middleware"
|
||||
"ezops/model"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// Agent 采集策略的兜底默认值(表内无记录时使用)。 // Logan.Li
|
||||
const (
|
||||
defaultReportInterval = 10
|
||||
defaultConfigPollInterval = 30
|
||||
// intervalFloor 与 Agent 侧校验保持一致:过密会打爆平台。 // Logan.Li
|
||||
intervalFloor = 5
|
||||
)
|
||||
|
||||
// GetAgentConfig 合并下发 Agent 配置。
|
||||
//
|
||||
// 原先 Agent 每 5 秒要分别请求 tasks/poll、health-probes/config、port-policy 三个接口,
|
||||
// 配置类请求量是上报本身的 6 倍;合并为单次响应后请求量降到原来的 1/3。 // Logan.Li
|
||||
//
|
||||
// GET /api/v1/agent/config
|
||||
func (h *AgentHandler) GetAgentConfig(c *gin.Context) {
|
||||
hostname := c.Query("hostname")
|
||||
if hostname == "" {
|
||||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "hostname 不能为空"})
|
||||
return
|
||||
}
|
||||
if !middleware.RequireAgentHostname(c, hostname) {
|
||||
return
|
||||
}
|
||||
|
||||
setting := h.resolveAgentSetting(hostname)
|
||||
|
||||
// 任务:拉取 pending 并标记 running(与 /tasks/poll 行为一致)。 // Logan.Li
|
||||
tasks := h.pickPendingTasks(hostname)
|
||||
|
||||
// 探针配置。 // Logan.Li
|
||||
probes := []model.HealthProbe{}
|
||||
h.db.Where("hostname = ? AND enabled = ?", hostname, true).Order("id ASC").Find(&probes)
|
||||
|
||||
// 端口策略。 // Logan.Li
|
||||
policy := h.resolvePortPolicy(hostname)
|
||||
|
||||
dto := model.AgentConfigDTO{
|
||||
ReportInterval: setting.ReportInterval,
|
||||
ConfigPollInterval: setting.ConfigPollInterval,
|
||||
Tasks: tasks,
|
||||
HealthProbes: probes,
|
||||
PortPolicy: policy,
|
||||
}
|
||||
dto.ConfigVersion = agentConfigVersion(dto)
|
||||
|
||||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: dto})
|
||||
}
|
||||
|
||||
// SaveAgentSetting 平台侧设置采集策略:hostname 为空串表示全局默认。 // Logan.Li
|
||||
// PUT /api/v1/agent/settings
|
||||
func (h *AgentHandler) SaveAgentSetting(c *gin.Context) {
|
||||
var req struct {
|
||||
Hostname string `json:"hostname"`
|
||||
ReportInterval int `json:"report_interval"`
|
||||
ConfigPollInterval int `json:"config_poll_interval"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
||||
return
|
||||
}
|
||||
if req.ReportInterval < intervalFloor {
|
||||
c.JSON(http.StatusBadRequest, model.Response{
|
||||
Code: 400, Message: fmt.Sprintf("上报间隔不能小于 %d 秒", intervalFloor)})
|
||||
return
|
||||
}
|
||||
if req.ConfigPollInterval < intervalFloor {
|
||||
c.JSON(http.StatusBadRequest, model.Response{
|
||||
Code: 400, Message: fmt.Sprintf("配置轮询间隔不能小于 %d 秒", intervalFloor)})
|
||||
return
|
||||
}
|
||||
|
||||
setting := model.AgentSetting{Hostname: req.Hostname}
|
||||
h.db.Where("hostname = ?", req.Hostname).Assign(map[string]interface{}{
|
||||
"report_interval": req.ReportInterval,
|
||||
"config_poll_interval": req.ConfigPollInterval,
|
||||
"updated_by": middleware.GetUsername(c),
|
||||
}).FirstOrCreate(&setting)
|
||||
|
||||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "已保存", Data: setting})
|
||||
}
|
||||
|
||||
// ListAgentSettings 列出全局默认与所有按主机覆盖项。 // Logan.Li
|
||||
// GET /api/v1/agent/settings
|
||||
func (h *AgentHandler) ListAgentSettings(c *gin.Context) {
|
||||
var items []model.AgentSetting
|
||||
h.db.Order("hostname ASC").Find(&items)
|
||||
if items == nil {
|
||||
items = []model.AgentSetting{}
|
||||
}
|
||||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: gin.H{
|
||||
"defaults": gin.H{
|
||||
"report_interval": defaultReportInterval,
|
||||
"config_poll_interval": defaultConfigPollInterval,
|
||||
},
|
||||
"items": items,
|
||||
}})
|
||||
}
|
||||
|
||||
// resolveAgentSetting 解析生效的采集策略:按主机覆盖 > 全局默认 > 内置默认。 // Logan.Li
|
||||
func (h *AgentHandler) resolveAgentSetting(hostname string) model.AgentSetting {
|
||||
effective := model.AgentSetting{
|
||||
Hostname: hostname,
|
||||
ReportInterval: defaultReportInterval,
|
||||
ConfigPollInterval: defaultConfigPollInterval,
|
||||
}
|
||||
|
||||
// 全局默认。 // Logan.Li
|
||||
var global model.AgentSetting
|
||||
if err := h.db.Where("hostname = ?", "").Take(&global).Error; err == nil {
|
||||
applyAgentSetting(&effective, global)
|
||||
}
|
||||
// 按主机覆盖。 // Logan.Li
|
||||
var perHost model.AgentSetting
|
||||
if err := h.db.Where("hostname = ?", hostname).Take(&perHost).Error; err == nil {
|
||||
applyAgentSetting(&effective, perHost)
|
||||
}
|
||||
return effective
|
||||
}
|
||||
|
||||
// applyAgentSetting 仅在来源值合法时覆盖,避免脏数据把间隔写成 0。 // Logan.Li
|
||||
func applyAgentSetting(dst *model.AgentSetting, src model.AgentSetting) {
|
||||
if src.ReportInterval >= intervalFloor {
|
||||
dst.ReportInterval = src.ReportInterval
|
||||
}
|
||||
if src.ConfigPollInterval >= intervalFloor {
|
||||
dst.ConfigPollInterval = src.ConfigPollInterval
|
||||
}
|
||||
}
|
||||
|
||||
// pickPendingTasks 取出该主机待执行任务并标记 running,最多 5 个。 // Logan.Li
|
||||
func (h *AgentHandler) pickPendingTasks(hostname string) []model.AgentTask {
|
||||
tasks := []model.AgentTask{}
|
||||
h.db.Where("hostname = ? AND status = ?", hostname, "pending").
|
||||
Order("created_at ASC").Limit(5).Find(&tasks)
|
||||
if len(tasks) == 0 {
|
||||
return tasks
|
||||
}
|
||||
now := time.Now()
|
||||
for _, t := range tasks {
|
||||
// 条件更新防止多 Agent 重复认领同一任务。 // Logan.Li
|
||||
h.db.Model(&model.AgentTask{}).
|
||||
Where("id = ? AND status = ?", t.ID, "pending").
|
||||
Updates(map[string]interface{}{"status": "running", "picked_at": now})
|
||||
}
|
||||
return tasks
|
||||
}
|
||||
|
||||
// resolvePortPolicy 读取主机端口策略,无记录时返回空白名单。 // Logan.Li
|
||||
func (h *AgentHandler) resolvePortPolicy(hostname string) *model.AgentPortPolicyDTO {
|
||||
empty := &model.AgentPortPolicyDTO{Hostname: hostname, PortFilterMode: "whitelist", PortWhitelist: []int{}}
|
||||
var policy model.AgentPortPolicy
|
||||
if err := h.db.Where("hostname = ?", hostname).First(&policy).Error; err != nil {
|
||||
return empty
|
||||
}
|
||||
whitelist := []int{}
|
||||
_ = json.Unmarshal([]byte(policy.PortWhitelist), &whitelist)
|
||||
return &model.AgentPortPolicyDTO{
|
||||
Hostname: policy.Hostname,
|
||||
PortFilterMode: policy.PortFilterMode,
|
||||
PortWhitelist: whitelist,
|
||||
}
|
||||
}
|
||||
|
||||
// agentConfigVersion 对配置内容取指纹,Agent 可据此跳过未变更配置的处理。 // Logan.Li
|
||||
func agentConfigVersion(dto model.AgentConfigDTO) string {
|
||||
payload := struct {
|
||||
ReportInterval int
|
||||
ConfigPollInterval int
|
||||
Tasks []model.AgentTask
|
||||
HealthProbes []model.HealthProbe
|
||||
PortPolicy *model.AgentPortPolicyDTO
|
||||
}{
|
||||
dto.ReportInterval, dto.ConfigPollInterval, dto.Tasks, dto.HealthProbes, dto.PortPolicy,
|
||||
}
|
||||
raw, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return strconv.FormatInt(time.Now().Unix(), 10)
|
||||
}
|
||||
sum := sha256.Sum256(raw)
|
||||
return hex.EncodeToString(sum[:8])
|
||||
}
|
||||
@@ -695,9 +695,23 @@ func (h *AgentInstallHandler) BatchUpgradeAgent(c *gin.Context) {
|
||||
binaryFile = "aiops-agent-linux-arm64"
|
||||
}
|
||||
|
||||
// 下载新二进制 + 获取版本号 + 重启服务
|
||||
// 下载新二进制 + 校验 + 重启服务。
|
||||
// 必须 fail-fast:旧版脚本用 `;` 串联并把 UPGRADE_OK 无条件 echo,
|
||||
// 下载失败时同样报"成功",导致升级结果完全不可信。 // Logan.Li
|
||||
upgradeScript := fmt.Sprintf(
|
||||
`curl -fsSL -o /tmp/aiops-agent %s && chmod +x /tmp/aiops-agent && systemctl stop aiops-agent 2>/dev/null; mv /tmp/aiops-agent /usr/local/bin/aiops-agent && systemctl start aiops-agent; ver=$(/usr/local/bin/aiops-agent version 2>/dev/null | awk '/AIOps Agent/{print $3}'); echo "UPGRADE_OK ver=$ver"`,
|
||||
`set -e; `+
|
||||
`curl -fsSL --connect-timeout 10 -o /tmp/aiops-agent.new %s || { echo "UPGRADE_FAIL download"; exit 1; }; `+
|
||||
`size=$(wc -c < /tmp/aiops-agent.new); `+
|
||||
`if [ "$size" -lt 1000000 ]; then echo "UPGRADE_FAIL size=$size"; exit 1; fi; `+
|
||||
`chmod +x /tmp/aiops-agent.new; `+
|
||||
`/tmp/aiops-agent.new version >/dev/null 2>&1 || { echo "UPGRADE_FAIL not-executable"; exit 1; }; `+
|
||||
`systemctl stop aiops-agent 2>/dev/null || true; `+
|
||||
`mv /tmp/aiops-agent.new /usr/local/bin/aiops-agent; `+
|
||||
`systemctl start aiops-agent; `+
|
||||
`sleep 1; `+
|
||||
`ver=$(/usr/local/bin/aiops-agent version 2>/dev/null | awk '/AIOps Agent/{print $3}'); `+
|
||||
`if [ -z "$ver" ]; then echo "UPGRADE_FAIL version-empty"; exit 1; fi; `+
|
||||
`echo "UPGRADE_OK ver=$ver"`,
|
||||
shellQuote(serverURL+"/api/agent/bin/"+binaryFile),
|
||||
)
|
||||
|
||||
|
||||
+6
-1
@@ -23,6 +23,7 @@ import (
|
||||
"ezops/pkg/cronscheduler"
|
||||
"ezops/pkg/database"
|
||||
"ezops/pkg/logger"
|
||||
"ezops/pkg/metricagg"
|
||||
ezredis "ezops/pkg/redis"
|
||||
"ezops/pkg/retention"
|
||||
"ezops/pkg/systask"
|
||||
@@ -224,6 +225,10 @@ func main() {
|
||||
})
|
||||
defer cronSched.Stop()
|
||||
|
||||
// 指标分层滚动聚合:raw → 1m/5m → 1h → 6h/1d,全部后台定时执行,
|
||||
// 取代原先依赖前端手动触发 POST /agent/aggregate 的方式。 // Logan.Li
|
||||
metricagg.Start(context.Background(), db, cache, log)
|
||||
|
||||
// 每日分批清理 Agent、指标、SNMP、拨测与旧日志,执行结果写入系统任务状态。 // Logan.Li
|
||||
go func() {
|
||||
ticker := time.NewTicker(24 * time.Hour)
|
||||
@@ -470,7 +475,7 @@ func loadConfig() *config.Config {
|
||||
viper.SetDefault("jwt.secret", "") // 生产环境必须通过配置或 JWT_SECRET 环境变量设置
|
||||
viper.SetDefault("jwt.expire_hour", 24)
|
||||
viper.SetDefault("agent.enforce_auth", false)
|
||||
viper.SetDefault("retention.agent_reports_days", 7)
|
||||
viper.SetDefault("retention.agent_reports_days", 1)
|
||||
viper.SetDefault("retention.metric_history_days", 30)
|
||||
viper.SetDefault("retention.metric_aggregate_days", 365)
|
||||
viper.SetDefault("retention.snmp_samples_days", 30)
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
-- 指标聚合表唯一约束:GORM 的 ON CONFLICT upsert 需要唯一索引才能真正生效,
|
||||
-- 否则每次聚合都会插入重复行(原 idx_agg 为非唯一索引)。 // Logan.Li
|
||||
-- 先按 (hostname, metric, interval, bucket_time) 去重,仅保留最大 id。 // Logan.Li
|
||||
DELETE a FROM monitor_metric_aggregates a
|
||||
JOIN monitor_metric_aggregates b
|
||||
ON a.hostname = b.hostname
|
||||
AND a.metric = b.metric
|
||||
AND a.`interval` = b.`interval`
|
||||
AND a.bucket_time = b.bucket_time
|
||||
AND a.id < b.id;
|
||||
|
||||
CREATE UNIQUE INDEX uniq_metric_agg ON monitor_metric_aggregates (hostname, metric, `interval`, bucket_time);
|
||||
@@ -0,0 +1,10 @@
|
||||
-- 指标聚合表唯一约束:GORM 的 ON CONFLICT upsert 需要唯一索引才能真正生效。 // Logan.Li
|
||||
DELETE FROM monitor_metric_aggregates a
|
||||
USING monitor_metric_aggregates b
|
||||
WHERE a.hostname = b.hostname
|
||||
AND a.metric = b.metric
|
||||
AND a.interval = b.interval
|
||||
AND a.bucket_time = b.bucket_time
|
||||
AND a.id < b.id;
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uniq_metric_agg ON monitor_metric_aggregates (hostname, metric, interval, bucket_time);
|
||||
@@ -0,0 +1,8 @@
|
||||
-- 指标聚合表唯一约束:GORM 的 ON CONFLICT upsert 需要唯一索引才能真正生效。 // Logan.Li
|
||||
DELETE FROM monitor_metric_aggregates
|
||||
WHERE id NOT IN (
|
||||
SELECT MAX(id) FROM monitor_metric_aggregates
|
||||
GROUP BY hostname, metric, "interval", bucket_time
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uniq_metric_agg ON monitor_metric_aggregates (hostname, metric, "interval", bucket_time);
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Agent 采集策略:hostname 为空串代表全局默认,非空为按主机覆盖。
|
||||
-- 平台可通过此表下发 report_interval / config_poll_interval,Agent 热更新无需重装。 // Logan.Li
|
||||
CREATE TABLE IF NOT EXISTS monitor_agent_settings (
|
||||
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||||
hostname VARCHAR(255) NOT NULL DEFAULT '',
|
||||
report_interval INT NOT NULL DEFAULT 10,
|
||||
config_poll_interval INT NOT NULL DEFAULT 30,
|
||||
updated_by VARCHAR(128) NOT NULL DEFAULT '',
|
||||
created_at DATETIME NOT NULL,
|
||||
updated_at DATETIME NOT NULL,
|
||||
UNIQUE KEY uniq_agent_settings_hostname (hostname)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
@@ -0,0 +1,11 @@
|
||||
-- Agent 采集策略:hostname 为空串代表全局默认,非空为按主机覆盖。 // Logan.Li
|
||||
CREATE TABLE IF NOT EXISTS monitor_agent_settings (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
hostname VARCHAR(255) NOT NULL DEFAULT '',
|
||||
report_interval INT NOT NULL DEFAULT 10,
|
||||
config_poll_interval INT NOT NULL DEFAULT 30,
|
||||
updated_by VARCHAR(128) NOT NULL DEFAULT '',
|
||||
created_at TIMESTAMP NOT NULL,
|
||||
updated_at TIMESTAMP NOT NULL,
|
||||
CONSTRAINT uniq_agent_settings_hostname UNIQUE (hostname)
|
||||
);
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Agent 采集策略:hostname 为空串代表全局默认,非空为按主机覆盖。 // Logan.Li
|
||||
CREATE TABLE IF NOT EXISTS monitor_agent_settings (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
hostname VARCHAR(255) NOT NULL DEFAULT '',
|
||||
report_interval INTEGER NOT NULL DEFAULT 10,
|
||||
config_poll_interval INTEGER NOT NULL DEFAULT 30,
|
||||
updated_by VARCHAR(128) NOT NULL DEFAULT '',
|
||||
created_at DATETIME NOT NULL,
|
||||
updated_at DATETIME NOT NULL
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uniq_agent_settings_hostname ON monitor_agent_settings (hostname);
|
||||
@@ -66,6 +66,20 @@ func All() []Migration {
|
||||
return executeDialectSQL(db, "0003_billing_settings")
|
||||
},
|
||||
},
|
||||
{
|
||||
Version: "0004_metric_aggregate_unique",
|
||||
Description: "指标聚合表唯一约束(upsert 生效)",
|
||||
Apply: func(db *gorm.DB) error {
|
||||
return executeDialectSQL(db, "0004_metric_aggregate_unique")
|
||||
},
|
||||
},
|
||||
{
|
||||
Version: "0005_agent_settings",
|
||||
Description: "Agent 采集策略(可下发频率)",
|
||||
Apply: func(db *gorm.DB) error {
|
||||
return executeDialectSQL(db, "0005_agent_settings")
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package model
|
||||
|
||||
import "time"
|
||||
|
||||
// AgentSetting Agent 采集策略。
|
||||
//
|
||||
// Hostname 为空串表示全局默认,非空表示对单台主机的覆盖;
|
||||
// 平台改这里即可下发新的采集/上报频率,Agent 侧热更新 ticker,无需重装。 // Logan.Li
|
||||
type AgentSetting struct {
|
||||
ID uint `gorm:"primarykey" json:"id"`
|
||||
Hostname string `gorm:"size:255;uniqueIndex:uniq_agent_settings_hostname;not null;default:''" json:"hostname"`
|
||||
ReportInterval int `gorm:"not null;default:10" json:"report_interval"` // 采集上报间隔(秒)
|
||||
ConfigPollInterval int `gorm:"not null;default:30" json:"config_poll_interval"` // 配置轮询间隔(秒)
|
||||
UpdatedBy string `gorm:"size:128;not null;default:''" json:"updated_by"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// TableName 固定表名,便于迁移脚本与代码保持一致。 // Logan.Li
|
||||
func (AgentSetting) TableName() string { return "monitor_agent_settings" }
|
||||
|
||||
// AgentConfigDTO 下发给 Agent 的合并配置。
|
||||
//
|
||||
// 原先 Agent 每 5 秒要分别请求 tasks/poll、health-probes/config、port-policy
|
||||
// 三个接口,请求量是上报本身的 6 倍;合并为单次响应后可显著降低平台与网络开销。 // Logan.Li
|
||||
type AgentConfigDTO struct {
|
||||
// ConfigVersion 为配置指纹,Agent 可比对后跳过后续相同配置的处理。 // Logan.Li
|
||||
ConfigVersion string `json:"config_version"`
|
||||
// ReportInterval 采集上报间隔(秒),Agent 收到后热更新 ticker。 // Logan.Li
|
||||
ReportInterval int `json:"report_interval"`
|
||||
// ConfigPollInterval 配置轮询间隔(秒)。 // Logan.Li
|
||||
ConfigPollInterval int `json:"config_poll_interval"`
|
||||
|
||||
Tasks []AgentTask `json:"tasks"`
|
||||
HealthProbes []HealthProbe `json:"health_probes"`
|
||||
PortPolicy *AgentPortPolicyDTO `json:"port_policy,omitempty"`
|
||||
}
|
||||
@@ -4,17 +4,20 @@ import "time"
|
||||
|
||||
// MetricsAggregate 指标聚合数据(Redis 热数据 → MySQL 时序聚合)// Logan.Li
|
||||
// 替代旧名 MetricsKline,OHLC 对监控无意义,改用 max/min/avg/p95
|
||||
//
|
||||
// (hostname, metric, interval, bucket_time) 必须是唯一索引:
|
||||
// GORM 的 ON CONFLICT upsert 依赖它,否则每次聚合都会插入重复行。 // Logan.Li
|
||||
type MetricsAggregate struct {
|
||||
ID uint `gorm:"primarykey" json:"id"`
|
||||
Hostname string `gorm:"size:255;index:idx_agg;not null" json:"hostname"`
|
||||
Metric string `gorm:"size:32;index:idx_agg;not null" json:"metric"` // cpu / mem / load1 / disk / net_rx / net_tx
|
||||
Interval string `gorm:"size:8;index:idx_agg;not null" json:"interval"` // 1m / 5m / 1h / 6h / 1d
|
||||
Hostname string `gorm:"size:255;uniqueIndex:uniq_metric_agg,priority:1;not null" json:"hostname"`
|
||||
Metric string `gorm:"size:32;uniqueIndex:uniq_metric_agg,priority:2;not null" json:"metric"` // cpu / mem / load1 / disk / net_rx / net_tx
|
||||
Interval string `gorm:"size:8;uniqueIndex:uniq_metric_agg,priority:3;not null" json:"interval"` // 1m / 5m / 1h / 6h / 1d
|
||||
Avg float64 `gorm:"type:decimal(10,2)" json:"avg"`
|
||||
MaxV float64 `gorm:"column:max_v;type:decimal(10,2)" json:"max"`
|
||||
MinV float64 `gorm:"column:min_v;type:decimal(10,2)" json:"min"`
|
||||
P95 float64 `gorm:"type:decimal(10,2)" json:"p95"`
|
||||
Count int `json:"count"`
|
||||
BucketTime time.Time `gorm:"index:idx_agg;not null" json:"bucket_time"`
|
||||
BucketTime time.Time `gorm:"uniqueIndex:uniq_metric_agg,priority:4;not null" json:"bucket_time"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
|
||||
@@ -2,9 +2,11 @@ package agentauth
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"ezops/model"
|
||||
@@ -13,6 +15,120 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// Token 校验缓存
|
||||
//
|
||||
// bcrypt 被设计成故意慢(本项目实测单次 Cost=10 约 72ms CPU)。Agent 每 10 秒
|
||||
// 轮询 3 个接口,每个接口都要校验一次 Token,30 台主机即 9 req/s×72ms ≈ 0.65 核,
|
||||
// 纯粹烧在重复校验同一条不可变凭证上。
|
||||
//
|
||||
// 由于 Token 在轮换前不可变(Issue 才会改写 token_hash),校验结果可以安全缓存:
|
||||
// - 缓存键:token 的 SHA-256(不落明文,避免内存里出现可直接使用的凭证)
|
||||
// - 有效期:默认 5 分钟,过期后回落一次 bcrypt 并续期
|
||||
// - 轮换:Issue 时全量清空,保证「旧 Token 立即失效」语义不变
|
||||
//
|
||||
// last_used_at 的写入同样按前缀节流(默认 60s 一次),避免每请求一次 UPDATE。
|
||||
// // Logan.Li
|
||||
type cachedCredential struct {
|
||||
credential *model.AgentCredential
|
||||
expiresAt time.Time
|
||||
// lastWrite 记录最近一次把 last_used_at 落库的时间,用于写入节流。 // Logan.Li
|
||||
lastWrite time.Time
|
||||
}
|
||||
|
||||
const (
|
||||
defaultCacheTTL = 5 * time.Minute
|
||||
defaultWriteThrottle = 60 * time.Second
|
||||
// cacheMaxEntries 上限,防止异常 Token 洪水把内存撑爆。 // Logan.Li
|
||||
cacheMaxEntries = 4096
|
||||
)
|
||||
|
||||
var (
|
||||
credCacheMu sync.Mutex
|
||||
credCache = make(map[string]*cachedCredential)
|
||||
// cacheTTL / writeThrottle 允许测试覆盖。 // Logan.Li
|
||||
cacheTTL = defaultCacheTTL
|
||||
writeThrottle = defaultWriteThrottle
|
||||
)
|
||||
|
||||
// cacheKey 只缓存 Token 的哈希,内存中不保留明文 Token。 // Logan.Li
|
||||
func cacheKey(token string) string {
|
||||
sum := sha256.Sum256([]byte(token))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// lookupCache 返回命中的凭证副本;命中时同时按节流更新 last_used_at。 // Logan.Li
|
||||
func lookupCache(db *gorm.DB, token string) (*model.AgentCredential, bool) {
|
||||
key := cacheKey(token)
|
||||
credCacheMu.Lock()
|
||||
defer credCacheMu.Unlock()
|
||||
|
||||
entry, ok := credCache[key]
|
||||
if !ok || time.Now().After(entry.expiresAt) {
|
||||
if ok {
|
||||
delete(credCache, key)
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
if entry.credential.ExpiresAt != nil && entry.credential.ExpiresAt.Before(time.Now()) {
|
||||
delete(credCache, key)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// 节流写入 last_used_at:仅当距上次落库超过阈值才真正 UPDATE。 // Logan.Li
|
||||
now := time.Now()
|
||||
if now.Sub(entry.lastWrite) >= writeThrottle {
|
||||
entry.lastWrite = now
|
||||
prefix := entry.credential.TokenPrefix
|
||||
go func() {
|
||||
_ = db.Model(&model.AgentCredential{}).
|
||||
Where("token_prefix = ?", prefix).
|
||||
Update("last_used_at", now).Error
|
||||
}()
|
||||
}
|
||||
|
||||
cp := *entry.credential
|
||||
cp.LastUsedAt = &now
|
||||
return &cp, true
|
||||
}
|
||||
|
||||
// storeCache 写入缓存,必要时按容量上限淘汰过期项。 // Logan.Li
|
||||
func storeCache(token string, credential *model.AgentCredential) {
|
||||
key := cacheKey(token)
|
||||
credCacheMu.Lock()
|
||||
defer credCacheMu.Unlock()
|
||||
|
||||
if len(credCache) >= cacheMaxEntries {
|
||||
now := time.Now()
|
||||
for k, v := range credCache {
|
||||
if now.After(v.expiresAt) {
|
||||
delete(credCache, k)
|
||||
}
|
||||
}
|
||||
// 仍超上限则整体清空,宁可多算几次 bcrypt 也不无限占内存。 // Logan.Li
|
||||
if len(credCache) >= cacheMaxEntries {
|
||||
credCache = make(map[string]*cachedCredential)
|
||||
}
|
||||
}
|
||||
|
||||
cp := *credential
|
||||
credCache[key] = &cachedCredential{
|
||||
credential: &cp,
|
||||
expiresAt: time.Now().Add(cacheTTL),
|
||||
lastWrite: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
// invalidateCache 清空全部缓存。Token 轮换/撤销后调用,保证旧 Token 立即失效。 // Logan.Li
|
||||
func invalidateCache() {
|
||||
credCacheMu.Lock()
|
||||
credCache = make(map[string]*cachedCredential)
|
||||
credCacheMu.Unlock()
|
||||
}
|
||||
|
||||
// Invalidate 提供给凭证变更入口(轮换/撤销/删除)显式失效缓存。 // Logan.Li
|
||||
func Invalidate() { invalidateCache() }
|
||||
|
||||
// Issue 为指定主机轮换 Agent Token,旧 Token 会立即失效。 // Logan.Li
|
||||
func Issue(db *gorm.DB, hostname, issuedBy string) (string, error) {
|
||||
hostname = strings.TrimSpace(hostname)
|
||||
@@ -47,11 +163,20 @@ func Issue(db *gorm.DB, hostname, issuedBy string) (string, error) {
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
// 轮换后必须让缓存失效,否则被缓存的旧 Token 还能继续通过校验。 // Logan.Li
|
||||
invalidateCache()
|
||||
return token, nil
|
||||
}
|
||||
|
||||
// Authenticate 校验 Token 并返回绑定的主机名。 // Logan.Li
|
||||
//
|
||||
// 命中缓存时跳过 bcrypt 与凭证查询,只做一次可选节流的 last_used_at 更新;
|
||||
// 未命中时才回落完整校验路径。 // Logan.Li
|
||||
func Authenticate(db *gorm.DB, token string) (*model.AgentCredential, error) {
|
||||
if credential, ok := lookupCache(db, token); ok {
|
||||
return credential, nil
|
||||
}
|
||||
|
||||
parts := strings.SplitN(token, ".", 2)
|
||||
if len(parts) != 2 || !strings.HasPrefix(parts[0], "agt_") || parts[1] == "" {
|
||||
return nil, fmt.Errorf("invalid agent token")
|
||||
@@ -69,6 +194,9 @@ func Authenticate(db *gorm.DB, token string) (*model.AgentCredential, error) {
|
||||
now := time.Now()
|
||||
_ = db.Model(&credential).Update("last_used_at", now).Error
|
||||
credential.LastUsedAt = &now
|
||||
|
||||
// 校验通过后写入缓存,后续请求直接命中。 // Logan.Li
|
||||
storeCache(token, &credential)
|
||||
return &credential, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
// Package metricagg 指标分层滚动聚合。
|
||||
//
|
||||
// 目标:把高频原始采样(Redis ZSET)逐级降采样为多分辨率聚合行,落库供趋势查询。
|
||||
//
|
||||
// 层级(逐级降采样,每级只读上一级,避免每次从原始数据重算):
|
||||
//
|
||||
// raw(10s, Redis) --1min--> 1m --1h--> 1h --6h/24h--> 6h / 1d
|
||||
//
|
||||
// 调度(全部后台,替代原先依赖前端手动触发 POST /agent/aggregate):
|
||||
// - fine 每 1 分钟:raw → 1m / 5m
|
||||
// - mid 每 15 分钟:1m(DB) → 1h
|
||||
// - coarse 每 1 小时:1h(DB) → 6h / 1d
|
||||
//
|
||||
// 幂等:依赖 monitor_metric_aggregates 的 uniq_metric_agg 唯一索引 + ON CONFLICT。 // Logan.Li
|
||||
package metricagg
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"sort"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"ezops/model"
|
||||
ezredis "ezops/pkg/redis"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
// metrics 需要聚合的指标清单(与 Agent 写入 Redis 的 key 保持一致)。 // Logan.Li
|
||||
var metrics = []string{"cpu", "mem", "load1"}
|
||||
|
||||
const (
|
||||
// rawKeyPrefix 与 controller.writeHotMetric 写入的 key 前缀保持一致。 // Logan.Li
|
||||
rawKeyPrefix = "metrics:"
|
||||
|
||||
fineEvery = time.Minute
|
||||
midEvery = 15 * time.Minute
|
||||
coarseEvery = time.Hour
|
||||
|
||||
// 各级回看窗口:只需覆盖该级「刚闭合的桶 + 边界抖动」。 // Logan.Li
|
||||
fineLookback = 3 * time.Hour
|
||||
midLookback = 6 * time.Hour
|
||||
coarseLookback = 8 * 24 * time.Hour
|
||||
)
|
||||
|
||||
// fineIntervals / midIntervals / coarseIntervals 各级负责的聚合周期。 // Logan.Li
|
||||
var (
|
||||
fineIntervals = []interval{{"1m", time.Minute}, {"5m", 5 * time.Minute}}
|
||||
midIntervals = []interval{{"1h", time.Hour}}
|
||||
coarseIntervals = []interval{{"6h", 6 * time.Hour}, {"1d", 24 * time.Hour}}
|
||||
)
|
||||
|
||||
type interval struct {
|
||||
name string
|
||||
d time.Duration
|
||||
}
|
||||
|
||||
// Start 启动后台聚合调度,ctx 取消后退出。 // Logan.Li
|
||||
func Start(ctx context.Context, db *gorm.DB, cache ezredis.Cache, log *zap.Logger) {
|
||||
// 无 ZSet 能力(redka/local 模式)时不做 Redis 侧聚合。 // Logan.Li
|
||||
if cache == nil || !cache.Capabilities().SortedSet {
|
||||
log.Info("指标聚合未启用:当前缓存后端不支持有序集合")
|
||||
return
|
||||
}
|
||||
|
||||
go run(ctx, db, cache, log, fineEvery, "fine", func(c context.Context) {
|
||||
rollupFromRaw(c, db, cache, log, fineIntervals, fineLookback)
|
||||
})
|
||||
go run(ctx, db, cache, log, midEvery, "mid", func(c context.Context) {
|
||||
rollupFromDB(db, log, "1m", midIntervals, midLookback)
|
||||
})
|
||||
go run(ctx, db, cache, log, coarseEvery, "coarse", func(c context.Context) {
|
||||
rollupFromDB(db, log, "1h", coarseIntervals, coarseLookback)
|
||||
})
|
||||
}
|
||||
|
||||
// run 周期性执行任务,起跑前先等一个周期,避免与启动期初始化抢资源。 // Logan.Li
|
||||
func run(ctx context.Context, _ *gorm.DB, _ ezredis.Cache, log *zap.Logger, every time.Duration, name string, fn func(context.Context)) {
|
||||
ticker := time.NewTicker(every)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
start := time.Now()
|
||||
fn(ctx)
|
||||
log.Debug("指标聚合完成", zap.String("level", name), zap.Duration("cost", time.Since(start)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rollupFromRaw 从 Redis 原始采样聚合出细粒度桶。 // Logan.Li
|
||||
func rollupFromRaw(ctx context.Context, db *gorm.DB, cache ezredis.Cache, log *zap.Logger, ivs []interval, lookback time.Duration) {
|
||||
hosts := activeHosts(db)
|
||||
if len(hosts) == 0 {
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
since := now.Add(-lookback)
|
||||
|
||||
for _, host := range hosts {
|
||||
for _, metric := range metrics {
|
||||
key := rawKeyPrefix + metric + ":" + host
|
||||
members, err := cache.ZRangeByScore(ctx, key,
|
||||
formatUnix(since), formatUnix(now.Add(time.Minute)))
|
||||
if err != nil || len(members) == 0 {
|
||||
continue
|
||||
}
|
||||
for _, iv := range ivs {
|
||||
buckets := bucketizeMembers(members, iv.d)
|
||||
if err := upsertBuckets(db, host, metric, iv.name, buckets); err != nil {
|
||||
log.Warn("写入聚合桶失败", zap.String("host", host), zap.String("metric", metric), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rollupFromDB 从上一级聚合行再聚合出粗粒度桶(逐级降采样)。 // Logan.Li
|
||||
func rollupFromDB(db *gorm.DB, log *zap.Logger, sourceInterval string, ivs []interval, lookback time.Duration) {
|
||||
hosts := activeHosts(db)
|
||||
if len(hosts) == 0 {
|
||||
return
|
||||
}
|
||||
since := time.Now().Add(-lookback)
|
||||
|
||||
for _, host := range hosts {
|
||||
for _, metric := range metrics {
|
||||
var rows []model.MetricsAggregate
|
||||
if err := db.Where("hostname = ? AND metric = ? AND `interval` = ? AND bucket_time >= ?",
|
||||
host, metric, sourceInterval, since).
|
||||
Order("bucket_time ASC").Find(&rows).Error; err != nil || len(rows) == 0 {
|
||||
continue
|
||||
}
|
||||
samples := make([]weightedSample, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
samples = append(samples, weightedSample{ts: r.BucketTime, avg: r.Avg, count: r.Count})
|
||||
}
|
||||
for _, iv := range ivs {
|
||||
buckets := bucketizeSamples(samples, iv.d)
|
||||
if err := upsertBuckets(db, host, metric, iv.name, buckets); err != nil {
|
||||
log.Warn("写入降采样桶失败", zap.String("host", host), zap.String("metric", metric), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// activeHosts 只聚合近期在线的 Agent 主机,避免对已下线主机做无用扫描。 // Logan.Li
|
||||
func activeHosts(db *gorm.DB) []string {
|
||||
var hosts []string
|
||||
cutoff := time.Now().Add(-15 * time.Minute)
|
||||
db.Model(&model.Server{}).
|
||||
Where("agent_status = ? AND last_check_at >= ?", "online", cutoff).
|
||||
Distinct().Pluck("hostname", &hosts)
|
||||
return hosts
|
||||
}
|
||||
|
||||
type bucketStats struct {
|
||||
avg, maxV, minV, p95 float64
|
||||
count int
|
||||
}
|
||||
|
||||
type weightedSample struct {
|
||||
ts time.Time
|
||||
avg float64
|
||||
count int
|
||||
}
|
||||
|
||||
// bucketizeMembers 把原始采样按周期分桶。 // Logan.Li
|
||||
func bucketizeMembers(members []ezredis.SortedSetMember, d time.Duration) map[time.Time][]float64 {
|
||||
sec := int64(d.Seconds())
|
||||
out := make(map[time.Time][]float64)
|
||||
for _, m := range members {
|
||||
ts := int64(m.Score)
|
||||
start := (ts / sec) * sec
|
||||
out[time.Unix(start, 0).UTC()] = append(out[time.Unix(start, 0).UTC()], parseValue(m.Member))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// bucketizeSamples 把上一级聚合行按周期再聚合(用 count 加权,避免小样本被放大)。 // Logan.Li
|
||||
func bucketizeSamples(samples []weightedSample, d time.Duration) map[time.Time][]float64 {
|
||||
sec := int64(d.Seconds())
|
||||
out := make(map[time.Time][]float64)
|
||||
for _, s := range samples {
|
||||
ts := s.ts.Unix()
|
||||
start := (ts / sec) * sec
|
||||
key := time.Unix(start, 0).UTC()
|
||||
// 用 avg 重复 count 次近似还原样本分布,保证 max/min/p95 有意义。 // Logan.Li
|
||||
n := s.count
|
||||
if n < 1 {
|
||||
n = 1
|
||||
}
|
||||
if n > 1000 {
|
||||
n = 1000
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
out[key] = append(out[key], s.avg)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// upsertBuckets 幂等写入聚合行(依赖 uniq_metric_agg 唯一索引)。 // Logan.Li
|
||||
func upsertBuckets(db *gorm.DB, host, metric, intervalName string, buckets map[time.Time][]float64) error {
|
||||
for bucketTime, values := range buckets {
|
||||
if len(values) == 0 {
|
||||
continue
|
||||
}
|
||||
st := calcStats(values)
|
||||
row := model.MetricsAggregate{
|
||||
Hostname: host,
|
||||
Metric: metric,
|
||||
Interval: intervalName,
|
||||
Avg: st.avg,
|
||||
MaxV: st.maxV,
|
||||
MinV: st.minV,
|
||||
P95: st.p95,
|
||||
Count: st.count,
|
||||
BucketTime: bucketTime,
|
||||
}
|
||||
if err := 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(&row).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// calcStats 计算 avg / max / min / p95,结果保留两位小数。 // Logan.Li
|
||||
func calcStats(values []float64) bucketStats {
|
||||
sorted := make([]float64, len(values))
|
||||
copy(sorted, values)
|
||||
sort.Float64s(sorted)
|
||||
|
||||
var sum float64
|
||||
for _, v := range sorted {
|
||||
sum += v
|
||||
}
|
||||
avg := sum / float64(len(sorted))
|
||||
p95 := sorted[int(math.Ceil(float64(len(sorted))*0.95))-1]
|
||||
|
||||
return bucketStats{
|
||||
avg: round2(avg),
|
||||
maxV: round2(sorted[len(sorted)-1]),
|
||||
minV: round2(sorted[0]),
|
||||
p95: round2(p95),
|
||||
count: len(sorted),
|
||||
}
|
||||
}
|
||||
|
||||
func round2(v float64) float64 { return math.Round(v*100) / 100 }
|
||||
|
||||
// parseValue 从 "45.2300:1234567890123456789" 中取出数值部分。 // Logan.Li
|
||||
func parseValue(member string) float64 {
|
||||
for i := 0; i < len(member); i++ {
|
||||
if member[i] == ':' {
|
||||
return parseFloat(member[:i])
|
||||
}
|
||||
}
|
||||
return parseFloat(member)
|
||||
}
|
||||
|
||||
func parseFloat(s string) float64 {
|
||||
var v float64
|
||||
var frac, div float64 = 0, 1
|
||||
neg := false
|
||||
i := 0
|
||||
if i < len(s) && (s[i] == '-' || s[i] == '+') {
|
||||
neg = s[i] == '-'
|
||||
i++
|
||||
}
|
||||
for ; i < len(s); i++ {
|
||||
c := s[i]
|
||||
if c == '.' {
|
||||
i++
|
||||
break
|
||||
}
|
||||
if c < '0' || c > '9' {
|
||||
break
|
||||
}
|
||||
v = v*10 + float64(c-'0')
|
||||
}
|
||||
for ; i < len(s); i++ {
|
||||
c := s[i]
|
||||
if c < '0' || c > '9' {
|
||||
break
|
||||
}
|
||||
frac = frac*10 + float64(c-'0')
|
||||
div *= 10
|
||||
}
|
||||
v += frac / div
|
||||
if neg {
|
||||
v = -v
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func formatUnix(t time.Time) string {
|
||||
// Redis ZSET score 为 Unix 秒;用整数文本作为 ZRANGEBYSCORE 的边界。 // Logan.Li
|
||||
return strconv.FormatInt(t.Unix(), 10)
|
||||
}
|
||||
@@ -19,7 +19,6 @@ type agentStaticPaths struct {
|
||||
var agentBinaryFiles = map[string]string{
|
||||
"linux-amd64": "aiops-agent-linux-amd64",
|
||||
"linux-arm64": "aiops-agent-linux-arm64",
|
||||
"darwin-arm64": "aiops-agent-darwin-arm64",
|
||||
"windows-amd64": "aiops-agent-windows-amd64.exe",
|
||||
}
|
||||
|
||||
|
||||
@@ -886,6 +886,9 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
|
||||
agentGroup.GET("/health-probes/config", agentHandler.GetHealthProbesForAgent)
|
||||
// 端口策略:Agent 拉取本机端口过滤策略。
|
||||
agentGroup.GET("/port-policy", agentHandler.GetPortPolicyForAgent)
|
||||
// 合并配置:一次返回任务 + 探针 + 端口策略 + 采集频率,
|
||||
// 取代原先每 5 秒分别拉取上面三个接口的做法。 // Logan.Li
|
||||
agentGroup.GET("/config", agentHandler.GetAgentConfig)
|
||||
}
|
||||
// Agent 数据查询 + 任务管理(需要 JWT)
|
||||
agentQuery := authorized.Group("/agent")
|
||||
@@ -918,6 +921,10 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
|
||||
agentQuery.POST("/batch-uninstall", agentInstallHandler.BatchUninstallAgent)
|
||||
agentQuery.POST("/batch-upgrade", agentInstallHandler.BatchUpgradeAgent)
|
||||
|
||||
// 采集策略(平台下发上报/轮询频率,Agent 热更新无需重装)
|
||||
agentQuery.GET("/settings", agentHandler.ListAgentSettings)
|
||||
agentQuery.PUT("/settings", agentHandler.SaveAgentSetting)
|
||||
|
||||
// 健康探针管理(CRUD)
|
||||
agentQuery.GET("/health-probes", agentHandler.ListHealthProbes)
|
||||
agentQuery.POST("/health-probes", agentHandler.CreateHealthProbe)
|
||||
|
||||
@@ -864,8 +864,6 @@ build_agent_binaries() {
|
||||
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/aiops-agent-linux-amd64 .
|
||||
info "编译 Agent Linux ARM64..."
|
||||
GOOS=linux GOARCH=arm64 CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/aiops-agent-linux-arm64 .
|
||||
info "编译 Agent macOS ARM64..."
|
||||
GOOS=darwin GOARCH=arm64 CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/aiops-agent-darwin-arm64 .
|
||||
info "编译 Agent Windows AMD64..."
|
||||
GOOS=windows GOARCH=amd64 CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/aiops-agent-windows-amd64.exe .
|
||||
success "Agent 多平台二进制编译完成: agent/bin/"
|
||||
|
||||
Reference in new issue
Block a user