diff --git a/go.mod b/go.mod index e5639fc..fd3aa5d 100644 --- a/go.mod +++ b/go.mod @@ -18,10 +18,12 @@ require ( github.com/jpillora/backoff v1.0.0 github.com/kardianos/service v1.2.4 github.com/labstack/echo/v4 v4.13.4 + github.com/minio/selfupdate v0.6.0 github.com/prometheus-community/pro-bing v0.7.0 github.com/shirou/gopsutil/v4 v4.25.10 github.com/spf13/afero v1.15.0 github.com/spf13/cobra v1.10.1 + github.com/valyala/fasttemplate v1.2.2 go.uber.org/zap v1.27.1 golang.org/x/crypto v0.45.0 golang.org/x/oauth2 v0.33.0 @@ -31,6 +33,7 @@ require ( ) require ( + aead.dev/minisign v0.2.0 // indirect filippo.io/edwards25519 v1.1.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/ebitengine/purego v0.9.0 // indirect @@ -66,7 +69,6 @@ require ( github.com/tklauser/go-sysconf v0.3.15 // indirect github.com/tklauser/numcpus v0.10.0 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect - github.com/valyala/fasttemplate v1.2.2 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect diff --git a/go.sum b/go.sum index 35b7af5..dc2cde3 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +aead.dev/minisign v0.2.0 h1:kAWrq/hBRu4AARY6AlciO83xhNnW9UaC8YipS2uhLPk= +aead.dev/minisign v0.2.0/go.mod h1:zdq6LdSd9TbuSxchxwhpA9zEb9YXcVGoE8JakuiGaIQ= filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= github.com/coreos/go-oidc/v3 v3.17.0 h1:hWBGaQfbi0iVviX4ibC7bk8OKT5qNr4klBaCHVNvehc= @@ -99,6 +101,8 @@ github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuE github.com/mattn/go-sqlite3 v1.14.32/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/microsoft/go-mssqldb v1.7.2 h1:CHkFJiObW7ItKTJfHo1QX7QBBD1iV+mn1eOyRP3b/PA= github.com/microsoft/go-mssqldb v1.7.2/go.mod h1:kOvZKUdrhhFQmxLZqbwUV0rHkNkZpthMITIb2Ko1IoA= +github.com/minio/selfupdate v0.6.0 h1:i76PgT0K5xO9+hjzKcacQtO7+MjJ4JKA8Ak8XQ9DDwU= +github.com/minio/selfupdate v0.6.0/go.mod h1:bO02GTIPCMQFTEvE5h4DjYB58bCoZ35XLeBf0buTDdM= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= @@ -154,27 +158,43 @@ go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc= go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= +golang.org/x/crypto v0.0.0-20211209193657-4570a0811e8b/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6 h1:zfMcR1Cs4KNuomFFgGefv5N0czO2XZpUbxGUy8i8ug0= golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6/go.mod h1:46edojNIoXTNOhySWIWdix628clX9ODXwPsQuG6hsK0= golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo= golang.org/x/oauth2 v0.33.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210228012217-479acdf4ea46/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0= golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ= golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/internal/app.go b/internal/app.go index 278792e..2bfb8e7 100644 --- a/internal/app.go +++ b/internal/app.go @@ -71,6 +71,11 @@ func setup(app *orz.App) error { app.Logger().Error("初始化默认属性配置失败", zap.Error(err)) // 不返回错误，继续启动 } + // 初始化探针的状态全部为离线 + if err := components.AgentService.InitStatus(ctx); err != nil { + app.Logger().Error("初始化探针状态失败", zap.Error(err)) + // 不返回错误，继续启动 + } // 启动WebSocket管理器 go components.WSManager.Run(ctx) diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index a4c8635..d8b7903 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -911,12 +911,7 @@ register_agent() { echo_info "正在注册探针..." - if /usr/local/bin/$AGENT_NAME register --endpoint "$endpoint" --token "$token"; then - echo_info "探针注册成功" - else - echo_error "探针注册失败" - exit 1 - fi + /usr/local/bin/$AGENT_NAME register --endpoint "$endpoint" --token "$token" } # 主流程 diff --git a/internal/models/alert.go b/internal/models/alert.go index acc6a0a..a0ad785 100644 --- a/internal/models/alert.go +++ b/internal/models/alert.go @@ -41,6 +41,10 @@ type AlertRules struct { // 服务下线告警配置 ServiceEnabled bool `json:"serviceEnabled"` // 是否启用服务下线告警 ServiceDuration int `json:"serviceDuration"` // 持续时间（秒） + + // 探针离线告警配置 + AgentOfflineEnabled bool `json:"agentOfflineEnabled"` // 是否启用探针离线告警 + AgentOfflineDuration int `json:"agentOfflineDuration"` // 持续时间（秒） } func (AlertConfig) TableName() string { diff --git a/internal/models/property.go b/internal/models/property.go index c1617a5..e74f10e 100644 --- a/internal/models/property.go +++ b/internal/models/property.go @@ -24,7 +24,22 @@ type NotificationChannelConfig struct { // dingtalk: { "secretKey": "xxx", "signSecret": "xxx" } // wecom: { "secretKey": "xxx" } // feishu: { "secretKey": "xxx", "signSecret": "xxx" } -// webhook: { "url": "https://..." } +// webhook: { +// "url": "https://...", +// "method": "POST", // 可选：GET, POST, PUT, PATCH, DELETE，默认 POST +// "headers": {"key": "value"}, // 可选：自定义请求头 +// "bodyTemplate": "json" // 可选：json(默认), form, custom +// "customBody": "" // 当 bodyTemplate 为 custom 时使用，支持变量替换 +// } + +// WebhookConfig 自定义 Webhook 配置结构 +type WebhookConfig struct { + URL string `json:"url"` // Webhook URL + Method string `json:"method,omitempty"` // 请求方法，默认 POST + Headers map[string]string `json:"headers,omitempty"` // 自定义请求头 + BodyTemplate string `json:"bodyTemplate,omitempty"` // 请求体模板：json, form, custom + CustomBody string `json:"customBody,omitempty"` // 自定义请求体模板（支持变量） +} type SystemConfig struct { SystemNameZh string `json:"systemNameZh"` // 系统名称（中文） diff --git a/internal/repo/agent_repo.go b/internal/repo/agent_repo.go index 29028e7..d49013e 100644 --- a/internal/repo/agent_repo.go +++ b/internal/repo/agent_repo.go @@ -2,7 +2,6 @@ package repo import ( "context" - "time" "github.com/dushixiang/pika/internal/models" "github.com/go-orz/orz" @@ -22,13 +21,13 @@ func NewAgentRepo(db *gorm.DB) *AgentRepo { } // UpdateStatus 更新探针状态 -func (r *AgentRepo) UpdateStatus(ctx context.Context, agentID string, status int) error { +func (r *AgentRepo) UpdateStatus(ctx context.Context, agentID string, status int, lastSeenAt int64) error { return r.db.WithContext(ctx). Model(&models.Agent{}). Where("id = ?", agentID). Updates(map[string]interface{}{ "status": status, - "last_seen_at": time.Now().UnixMilli(), + "last_seen_at": lastSeenAt, }).Error } diff --git a/internal/service/agent_service.go b/internal/service/agent_service.go index 14f0fed..e4708fe 100644 --- a/internal/service/agent_service.go +++ b/internal/service/agent_service.go @@ -109,7 +109,7 @@ func (s *AgentService) RegisterAgent(ctx context.Context, ip string, info *proto // UpdateAgentStatus 更新探针状态 func (s *AgentService) UpdateAgentStatus(ctx context.Context, agentID string, status int) error { - return s.AgentRepo.UpdateStatus(ctx, agentID, status) + return s.AgentRepo.UpdateStatus(ctx, agentID, status, time.Now().UnixMilli()) } // GetAgent 获取探针信息 @@ -790,3 +790,16 @@ func (s *AgentService) GetAgentByAuth(ctx context.Context, id string, isAuthenti func (s *AgentService) GetAllTags(ctx context.Context) ([]string, error) { return s.AgentRepo.GetAllTags(ctx) } + +func (s *AgentService) InitStatus(ctx context.Context) error { + agents, err := s.AgentRepo.FindAll(ctx) + if err != nil { + return err + } + for _, agent := range agents { + if err := s.AgentRepo.UpdateStatus(ctx, agent.ID, 0, 0); err != nil { + return err + } + } + return nil +} diff --git a/internal/service/alert_service.go b/internal/service/alert_service.go index 8d7e7d5..cb23a71 100644 --- a/internal/service/alert_service.go +++ b/internal/service/alert_service.go @@ -354,6 +354,13 @@ func (s *AlertService) CheckMonitorAlerts(ctx context.Context) error { s.logger.Error("检查服务下线告警失败", zap.Error(err)) } } + + // 检查探针离线告警 + if config.Rules.AgentOfflineEnabled { + if err := s.checkAgentOfflineAlerts(ctx, &config, now); err != nil { + s.logger.Error("检查探针离线告警失败", zap.Error(err)) + } + } } return nil @@ -710,3 +717,152 @@ func (s *AlertService) resolveServiceDownAlert(ctx context.Context, config *mode state.IsFiring = false state.LastRecordID = 0 } + +// checkAgentOfflineAlerts 检查探针离线告警 +func (s *AlertService) checkAgentOfflineAlerts(ctx context.Context, config *models.AlertConfig, now int64) error { + // 获取所有探针 + agents, err := s.agentRepo.FindAll(ctx) + if err != nil { + return err + } + + for _, agent := range agents { + stateKey := fmt.Sprintf("%s:%s:agent_offline:%s", config.AgentID, config.ID, agent.ID) + + s.mu.Lock() + state, exists := s.states[stateKey] + if !exists { + state = &models.AlertState{ + AgentID: agent.ID, + ConfigID: config.ID, + AlertType: "agent_offline", + Threshold: 0, + Duration: config.Rules.AgentOfflineDuration, + } + s.states[stateKey] = state + } + s.mu.Unlock() + + state.LastCheckTime = now + + // 计算探针离线时间（秒） + offlineSeconds := (now - agent.LastSeenAt) / 1000 + + // 检查探针是否离线超过阈值 + if offlineSeconds >= int64(config.Rules.AgentOfflineDuration) { + // 探针离线超过阈值 + if !state.IsFiring { + // 触发告警 + s.fireAgentOfflineAlert(ctx, config, &agent, state, offlineSeconds, now) + } + } else { + // 探针在线或离线时间未达到阈值 + if state.IsFiring { + // 从告警状态变为恢复状态 + s.resolveAgentOfflineAlert(ctx, config, &agent, state) + } + } + } + + return nil +} + +// fireAgentOfflineAlert 触发探针离线告警 +func (s *AlertService) fireAgentOfflineAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, state *models.AlertState, offlineSeconds int64, now int64) { + s.logger.Info("触发探针离线告警", + zap.String("agentId", agent.ID), + zap.String("agentName", agent.Name), + zap.Int64("offlineSeconds", offlineSeconds), + zap.Int("threshold", state.Duration), + ) + + // 创建告警记录 + record := &models.AlertRecord{ + AgentID: agent.ID, + ConfigID: config.ID, + ConfigName: config.Name, + AlertType: "agent_offline", + Message: fmt.Sprintf("探针 %s 已离线%d秒，超过阈值%d秒", agent.Name, offlineSeconds, state.Duration), + Threshold: float64(state.Duration), + ActualValue: float64(offlineSeconds), + Level: "critical", + Status: "firing", + FiredAt: now, + CreatedAt: now, + } + + err := s.alertRepo.CreateAlertRecord(ctx, record) + if err != nil { + s.logger.Error("创建探针离线告警记录失败", zap.Error(err)) + return + } + + state.IsFiring = true + state.LastRecordID = record.ID + + // 发送通知 + go func() { + channelConfigs, err := s.propertyService.GetNotificationChannelConfigs(context.Background()) + if err != nil { + s.logger.Error("获取通知渠道配置失败", zap.Error(err)) + return + } + + var enabledChannels []models.NotificationChannelConfig + for _, channel := range channelConfigs { + if channel.Enabled { + enabledChannels = append(enabledChannels, channel) + } + } + + if err := s.notifier.SendNotificationByConfigs(context.Background(), enabledChannels, record, agent); err != nil { + s.logger.Error("发送探针离线告警通知失败", zap.Error(err)) + } + }() +} + +// resolveAgentOfflineAlert 恢复探针离线告警 +func (s *AlertService) resolveAgentOfflineAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, state *models.AlertState) { + s.logger.Info("探针离线告警恢复", + zap.String("agentId", agent.ID), + zap.String("agentName", agent.Name), + ) + + // 更新告警记录状态 + if state.LastRecordID > 0 { + existingRecord, err := s.alertRepo.GetLatestAlertRecord(ctx, config.ID, "agent_offline") + if err == nil && existingRecord != nil { + existingRecord.Status = "resolved" + existingRecord.ResolvedAt = time.Now().UnixMilli() + existingRecord.UpdatedAt = time.Now().UnixMilli() + + err = s.alertRepo.UpdateAlertRecord(ctx, existingRecord) + if err != nil { + s.logger.Error("更新探针离线告警记录失败", zap.Error(err)) + } else { + // 发送恢复通知 + go func() { + channelConfigs, err := s.propertyService.GetNotificationChannelConfigs(context.Background()) + if err != nil { + s.logger.Error("获取通知渠道配置失败", zap.Error(err)) + return + } + + var enabledChannels []models.NotificationChannelConfig + for _, channel := range channelConfigs { + if channel.Enabled { + enabledChannels = append(enabledChannels, channel) + } + } + + if err := s.notifier.SendNotificationByConfigs(context.Background(), enabledChannels, existingRecord, agent); err != nil { + s.logger.Error("发送探针恢复通知失败", zap.Error(err)) + } + }() + } + } + } + + state.IsFiring = false + state.LastRecordID = 0 +} diff --git a/internal/service/notifier.go b/internal/service/notifier.go index a52297c..6356caf 100644 --- a/internal/service/notifier.go +++ b/internal/service/notifier.go @@ -10,9 +10,12 @@ import ( "fmt" "io" "net/http" + "net/url" + "strings" "time" "github.com/dushixiang/pika/internal/models" + "github.com/valyala/fasttemplate" "go.uber.org/zap" ) @@ -185,19 +188,182 @@ func (n *Notifier) sendFeishu(ctx context.Context, webhook, message string) erro } // sendCustomWebhook 发送自定义Webhook -func (n *Notifier) sendCustomWebhook(ctx context.Context, webhook string, message string) error { - // 发送完整的告警记录和探针信息 - body := map[string]interface{}{ - "msg_type": "text", - "text": map[string]string{ - "content": message, - }, +func (n *Notifier) sendCustomWebhook(ctx context.Context, config map[string]interface{}, agent *models.Agent, record *models.AlertRecord) error { + // 解析配置 + webhookURL, ok := config["url"].(string) + if !ok || webhookURL == "" { + return fmt.Errorf("自定义Webhook配置缺少 url") } - _, err := n.sendJSONRequest(ctx, webhook, body) - if err != nil { - return err + // 获取请求方法，默认 POST + method := "POST" + if m, ok := config["method"].(string); ok && m != "" { + method = strings.ToUpper(m) } + + // 获取自定义请求头 + headers := make(map[string]string) + if h, ok := config["headers"].(map[string]interface{}); ok { + for k, v := range h { + if strVal, ok := v.(string); ok { + headers[k] = strVal + } + } + } + + // 获取请求体模板类型，默认 json + bodyTemplate := "json" + if bt, ok := config["bodyTemplate"].(string); ok && bt != "" { + bodyTemplate = bt + } + + // 构建消息内容 + message := n.buildMessage(agent, record) + + // 根据模板类型构建请求体 + var reqBody io.Reader + var contentType string + + switch bodyTemplate { + case "json": + // JSON 格式 + body := map[string]interface{}{ + "msg_type": "text", + "text": map[string]string{ + "content": message, + }, + "agent": map[string]interface{}{ + "id": agent.ID, + "name": agent.Name, + "hostname": agent.Hostname, + "ip": agent.IP, + }, + "alert": map[string]interface{}{ + "type": record.AlertType, + "level": record.Level, + "status": record.Status, + "message": record.Message, + "threshold": record.Threshold, + "actualValue": record.ActualValue, + "firedAt": record.FiredAt, + "resolvedAt": record.ResolvedAt, + }, + } + data, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("序列化 JSON 失败: %w", err) + } + reqBody = bytes.NewReader(data) + contentType = "application/json" + + case "form": + // Form 表单格式 + formData := url.Values{} + formData.Set("message", message) + formData.Set("agent_id", agent.ID) + formData.Set("agent_name", agent.Name) + formData.Set("agent_hostname", agent.Hostname) + formData.Set("agent_ip", agent.IP) + formData.Set("alert_type", record.AlertType) + formData.Set("alert_level", record.Level) + formData.Set("alert_status", record.Status) + formData.Set("alert_message", record.Message) + formData.Set("threshold", fmt.Sprintf("%.2f", record.Threshold)) + formData.Set("actual_value", fmt.Sprintf("%.2f", record.ActualValue)) + formData.Set("fired_at", fmt.Sprintf("%d", record.FiredAt)) + if record.ResolvedAt > 0 { + formData.Set("resolved_at", fmt.Sprintf("%d", record.ResolvedAt)) + } + reqBody = strings.NewReader(formData.Encode()) + contentType = "application/x-www-form-urlencoded" + + case "custom": + // 自定义模板，支持变量替换 + customBody, ok := config["customBody"].(string) + if !ok || customBody == "" { + return fmt.Errorf("使用 custom 模板时必须提供 customBody") + } + + // 使用 fasttemplate 进行变量替换 + t := fasttemplate.New(customBody, "{{", "}}") + bodyStr := t.ExecuteFuncString(func(w io.Writer, tag string) (int, error) { + switch tag { + case "message": + return w.Write([]byte(message)) + case "agent.id": + return w.Write([]byte(agent.ID)) + case "agent.name": + return w.Write([]byte(agent.Name)) + case "agent.hostname": + return w.Write([]byte(agent.Hostname)) + case "agent.ip": + return w.Write([]byte(agent.IP)) + case "alert.type": + return w.Write([]byte(record.AlertType)) + case "alert.level": + return w.Write([]byte(record.Level)) + case "alert.status": + return w.Write([]byte(record.Status)) + case "alert.message": + return w.Write([]byte(record.Message)) + case "alert.threshold": + return w.Write([]byte(fmt.Sprintf("%.2f", record.Threshold))) + case "alert.actualValue": + return w.Write([]byte(fmt.Sprintf("%.2f", record.ActualValue))) + case "alert.firedAt": + return w.Write([]byte(fmt.Sprintf("%d", record.FiredAt))) + case "alert.resolvedAt": + return w.Write([]byte(fmt.Sprintf("%d", record.ResolvedAt))) + default: + // 未知变量保持原样 + return w.Write([]byte("{{" + tag + "}}")) + } + }) + reqBody = strings.NewReader(bodyStr) + contentType = "text/plain" + + default: + return fmt.Errorf("不支持的 bodyTemplate: %s", bodyTemplate) + } + + // 创建请求 + req, err := http.NewRequestWithContext(ctx, method, webhookURL, reqBody) + if err != nil { + return fmt.Errorf("创建请求失败: %w", err) + } + + // 设置 Content-Type + req.Header.Set("Content-Type", contentType) + + // 设置自定义请求头 + for k, v := range headers { + req.Header.Set(k, v) + } + + // 发送请求 + client := &http.Client{ + Timeout: 10 * time.Second, + } + + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("发送请求失败: %w", err) + } + defer resp.Body.Close() + + // 读取响应 + respBody, _ := io.ReadAll(resp.Body) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("请求失败，状态码: %d, 响应: %s", resp.StatusCode, string(respBody)) + } + + n.logger.Info("自定义Webhook发送成功", + zap.String("url", webhookURL), + zap.String("method", method), + zap.String("response", string(respBody)), + ) + return nil } @@ -279,13 +445,8 @@ func (n *Notifier) sendFeishuByConfig(ctx context.Context, config map[string]int } // sendWebhookByConfig 根据配置发送自定义Webhook -func (n *Notifier) sendWebhookByConfig(ctx context.Context, config map[string]interface{}, message string) error { - url, ok := config["url"].(string) - if !ok || url == "" { - return fmt.Errorf("自定义Webhook配置缺少 url") - } - - return n.sendCustomWebhook(ctx, url, message) +func (n *Notifier) sendWebhookByConfig(ctx context.Context, config map[string]interface{}, agent *models.Agent, record *models.AlertRecord) error { + return n.sendCustomWebhook(ctx, config, agent, record) } // SendNotificationByConfig 根据新的配置结构发送通知 @@ -309,7 +470,7 @@ func (n *Notifier) SendNotificationByConfig(ctx context.Context, channelConfig * case "feishu": return n.sendFeishuByConfig(ctx, channelConfig.Config, message) case "webhook": - return n.sendWebhookByConfig(ctx, channelConfig.Config, message) + return n.sendWebhookByConfig(ctx, channelConfig.Config, agent, record) case "email": // TODO: 实现邮件通知 return fmt.Errorf("邮件通知暂未实现") @@ -354,7 +515,23 @@ func (n *Notifier) SendFeishuByConfig(ctx context.Context, config map[string]int return n.sendFeishuByConfig(ctx, config, message) } -// SendWebhookByConfig 导出方法供外部调用 +// SendWebhookByConfig 导出方法供外部调用（测试用） func (n *Notifier) SendWebhookByConfig(ctx context.Context, config map[string]interface{}, message string) error { - return n.sendWebhookByConfig(ctx, config, message) + // 为了测试，创建一个临时的 agent 和 record + agent := &models.Agent{ + ID: "test-agent", + Name: "测试探针", + Hostname: "test-host", + IP: "127.0.0.1", + } + record := &models.AlertRecord{ + AlertType: "test", + Level: "info", + Status: "firing", + Message: message, + Threshold: 0, + ActualValue: 0, + FiredAt: time.Now().UnixMilli(), + } + return n.sendWebhookByConfig(ctx, config, agent, record) } diff --git a/pkg/agent/updater/updater.go b/pkg/agent/updater/updater.go index 32a6b05..225779c 100644 --- a/pkg/agent/updater/updater.go +++ b/pkg/agent/updater/updater.go @@ -5,16 +5,14 @@ import ( "crypto/tls" "encoding/json" "fmt" - "io" "log" "net/http" "os" - "os/exec" - "path/filepath" "runtime" "time" "github.com/dushixiang/pika/pkg/agent/config" + "github.com/minio/selfupdate" ) // VersionInfo 版本信息 @@ -145,23 +143,6 @@ func (u *Updater) downloadAndUpdate(versionInfo *VersionInfo) error { log.Printf("📥 下载新版本: %s", versionInfo.Version) downloadURL := u.cfg.GetDownloadURL() - if err := u.updateWithClient(downloadURL); err != nil { - return err - } - - log.Printf("✅ 新版本已安装到: %s", u.executablePath) - return nil -} - -// updateWithClient 使用实例的 httpClient 下载并更新 -func (u *Updater) updateWithClient(downloadURL string) error { - execPath := u.executablePath - - // 解析实际路径（处理符号链接） - execPath, err := filepath.EvalSymlinks(execPath) - if err != nil { - return fmt.Errorf("解析可执行文件路径失败: %w", err) - } // 下载文件 resp, err := u.httpClient.Get(downloadURL) @@ -174,31 +155,18 @@ func (u *Updater) updateWithClient(downloadURL string) error { return fmt.Errorf("HTTP 状态码: %d", resp.StatusCode) } - // 创建临时文件 - tmpFile := execPath + ".new" - out, err := os.OpenFile(tmpFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0755) - if err != nil { - return fmt.Errorf("创建临时文件失败: %w", err) - } - defer out.Close() - defer os.Remove(tmpFile) - - // 写入文件 - written, err := io.Copy(out, resp.Body) - if err != nil { - return fmt.Errorf("写入文件失败: %w", err) + // 使用 selfupdate 应用更新 + if err := selfupdate.Apply(resp.Body, selfupdate.Options{}); err != nil { + return fmt.Errorf("应用更新失败: %w", err) } - log.Printf("下载完成，文件大小: %d 字节", written) + log.Printf("✅ 更新成功，进程即将退出，等待系统服务重启...") - // 根据操作系统选择不同的更新策略 - if runtime.GOOS == "windows" { - // Windows: 使用批处理脚本延迟替换 - return updateOnWindows(execPath, tmpFile) - } + // 退出当前进程，让系统服务管理器（systemd/supervisor等）自动重启 + // 注意：这要求服务配置了自动重启（如 systemd 的 Restart=always） + os.Exit(1) - // Unix-like: 直接替换 - return updateOnUnix(execPath, tmpFile) + return nil } // CheckUpdate 手动检查更新（用于命令行） @@ -229,17 +197,6 @@ func CheckUpdate(updateURL, currentVer string) (*VersionInfo, error) { // Update 手动更新（用于命令行） func Update(downloadURL string) error { - execPath, err := os.Executable() - if err != nil { - return fmt.Errorf("获取可执行文件路径失败: %w", err) - } - - // 解析实际路径（处理符号链接） - execPath, err = filepath.EvalSymlinks(execPath) - if err != nil { - return fmt.Errorf("解析可执行文件路径失败: %w", err) - } - client := &http.Client{ Timeout: 300 * time.Second, } @@ -255,52 +212,12 @@ func Update(downloadURL string) error { return fmt.Errorf("HTTP 状态码: %d", resp.StatusCode) } - // 创建临时文件 - tmpFile := execPath + ".new" - out, err := os.OpenFile(tmpFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0755) - if err != nil { - return fmt.Errorf("创建临时文件失败: %w", err) - } - defer out.Close() - defer os.Remove(tmpFile) - - // 写入文件 - written, err := io.Copy(out, resp.Body) - if err != nil { - return fmt.Errorf("写入文件失败: %w", err) + // 使用 selfupdate 应用更新 + if err := selfupdate.Apply(resp.Body, selfupdate.Options{}); err != nil { + return fmt.Errorf("应用更新失败: %w", err) } - log.Printf("下载完成，文件大小: %d 字节", written) - - // 根据操作系统选择不同的更新策略 - if runtime.GOOS == "windows" { - // Windows: 使用批处理脚本延迟替换 - return updateOnWindows(execPath, tmpFile) - } - - // Unix-like: 直接替换 - return updateOnUnix(execPath, tmpFile) -} - -// updateOnUnix Unix-like 系统的更新逻辑 -func updateOnUnix(execPath, tmpFile string) error { - // 备份旧版本 - backupPath := execPath + ".bak" - if err := os.Rename(execPath, backupPath); err != nil { - return fmt.Errorf("备份旧版本失败: %w", err) - } - - // 替换为新版本 - if err := os.Rename(tmpFile, execPath); err != nil { - // 恢复备份 - os.Rename(backupPath, execPath) - return fmt.Errorf("替换新版本失败: %w", err) - } - - // 删除备份 - os.Remove(backupPath) - - log.Println("✅ 更新完成，进程即将退出，等待系统服务重启...") + log.Printf("✅ 更新成功，进程即将退出，等待系统服务重启...") // 退出当前进程，让系统服务管理器（systemd/supervisor等）自动重启 // 注意：这要求服务配置了自动重启（如 systemd 的 Restart=always） @@ -308,42 +225,3 @@ func updateOnUnix(execPath, tmpFile string) error { return nil } - -// updateOnWindows Windows 系统的更新逻辑 -func updateOnWindows(execPath, tmpFile string) error { - // 在 Windows 上，无法直接替换正在运行的可执行文件 - // 策略: 创建一个批处理脚本来延迟替换和重启 - - batScript := execPath + ".update.bat" - batContent := fmt.Sprintf(`@echo off -timeout /t 2 /nobreak >nul -move /y "%s" "%s.bak" -move /y "%s" "%s" -del "%s.bak" -start "" "%s" -del "%%~f0" -`, execPath, execPath, tmpFile, execPath, execPath, execPath) - - if err := os.WriteFile(batScript, []byte(batContent), 0755); err != nil { - return fmt.Errorf("创建更新脚本失败: %w", err) - } - - log.Println("✅ 更新完成，准备重启进程...") - - // 启动批处理脚本并退出当前进程 - cmd := exec.Command("cmd.exe", "/C", batScript) - cmd.Stdout = nil - cmd.Stderr = nil - cmd.Stdin = nil - - if err := cmd.Start(); err != nil { - os.Remove(batScript) - return fmt.Errorf("启动更新脚本失败: %w", err) - } - - // 让系统服务管理器来重启（当前进程退出后） - log.Println("进程即将退出，等待系统服务重启...") - os.Exit(0) - - return nil -} diff --git a/web/src/components/PublicFooter.tsx b/web/src/components/PublicFooter.tsx index f57932b..64da96f 100644 --- a/web/src/components/PublicFooter.tsx +++ b/web/src/components/PublicFooter.tsx @@ -5,12 +5,12 @@ const PublicFooter = () => { const icpCode = window.SystemConfig?.ICPCode || ''; return ( -