From a7d1795463205927649cf894bd146ce19190a392 Mon Sep 17 00:00:00 2001 From: dushixiang Date: Thu, 13 Nov 2025 20:09:06 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/agent/main.go | 5 - internal/app.go | 53 ++ internal/config/config.go | 3 +- internal/handler/agent_handler.go | 52 +- internal/handler/api_key_handler.go | 2 +- internal/handler/monitor_handler.go | 162 ++++++ internal/handler/user_handler.go | 2 +- internal/models/metric.go | 26 +- internal/models/monitor_task.go | 26 + internal/protocol/message.go | 6 +- internal/protocol/monitor.go | 31 ++ internal/providers.go | 80 --- internal/repo/agent_repo.go | 12 + internal/repo/metric_repo.go | 166 ++++--- internal/repo/monitor_repo.go | 31 ++ internal/scheduler/monitor_scheduler.go | 283 +++++++++++ internal/service/account_service.go | 6 +- internal/service/agent_service.go | 112 ++--- internal/service/alert_service.go | 7 +- internal/service/api_key_service.go | 27 +- internal/service/monitor_service.go | 477 ++++++++++++++++++ internal/service/property_service.go | 5 +- internal/service/user_service.go | 39 +- internal/wire.go | 43 +- internal/wire_gen.go | 41 +- pkg/agent/collector/docker.go | 38 +- pkg/agent/collector/manager.go | 11 +- pkg/agent/collector/monitor.go | 41 +- pkg/agent/config/config.go | 75 --- pkg/agent/service/agent.go | 135 ++--- web/src/api/agent.ts | 3 +- web/src/api/monitor.ts | 57 +++ web/src/components/PublicFooter.tsx | 11 + web/src/components/PublicHeader.tsx | 119 +++++ web/src/pages/AdminLayout/index.tsx | 24 +- web/src/pages/Agents/AgentList.tsx | 14 +- web/src/pages/Monitors/MonitorList.tsx | 492 +++++++++++++++++++ web/src/pages/Public/MonitorDetail.tsx | 620 +++++++++++++++++++++++ web/src/pages/Public/MonitorList.tsx | 475 ++++++++++++++++++ web/src/pages/Public/ServerDetail.tsx | 624 ++++++++++-------------- web/src/pages/Public/ServerList.tsx | 110 +---- web/src/router/index.tsx | 15 + web/src/types/index.ts | 118 ++++- 43 files changed, 3683 insertions(+), 996 deletions(-) create mode 100644 internal/handler/monitor_handler.go create mode 100644 internal/models/monitor_task.go create mode 100644 internal/protocol/monitor.go delete mode 100644 internal/providers.go create mode 100644 internal/repo/monitor_repo.go create mode 100644 internal/scheduler/monitor_scheduler.go create mode 100644 internal/service/monitor_service.go create mode 100644 web/src/api/monitor.ts create mode 100644 web/src/components/PublicFooter.tsx create mode 100644 web/src/components/PublicHeader.tsx create mode 100644 web/src/pages/Monitors/MonitorList.tsx create mode 100644 web/src/pages/Public/MonitorDetail.tsx create mode 100644 web/src/pages/Public/MonitorList.tsx diff --git a/cmd/agent/main.go b/cmd/agent/main.go index e6dfca1..c30e5b1 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -495,11 +495,6 @@ func registerAgent(cmd *cobra.Command, args []string) { Enabled: true, CheckInterval: "10m", }, - Monitor: config.MonitorConfig{ - Enabled: false, - Interval: 60, - Items: []config.MonitorItem{}, - }, } // 6. 保存配置 diff --git a/internal/app.go b/internal/app.go index 015d275..599e63c 100644 --- a/internal/app.go +++ b/internal/app.go @@ -10,6 +10,7 @@ import ( "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/handler" "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/scheduler" "github.com/dushixiang/pika/internal/service" "github.com/dushixiang/pika/web" "github.com/google/uuid" @@ -80,6 +81,13 @@ func setup(app *orz.App) error { // 启动指标监控任务（用于告警检测） go startMetricsMonitoring(ctx, components, app.Logger()) + // 启动服务监控任务调度器 + monitorScheduler := scheduler.NewMonitorScheduler(components.MonitorService, app.Logger(), 10) + monitorScheduler.Start(ctx) + + // 启动监控统计计算任务 + go startMonitorStatsCalculation(ctx, components, app.Logger()) + // 设置API setupApi(app, components) @@ -134,6 +142,11 @@ func setupApi(app *orz.App, components *AppComponents) { // Agent 版本和下载 publicApi.GET("/agent/version", components.AgentHandler.GetAgentVersion) publicApi.GET("/agent/downloads/:filename", components.AgentHandler.DownloadAgent) + + // 监控统计数据（公开访问）- 用于公共展示页面 + publicApi.GET("/monitors/stats", components.MonitorHandler.GetAllStats) + publicApi.GET("/monitors/:name/stats", components.MonitorHandler.GetStatsByName) + publicApi.GET("/monitors/:name/history", components.MonitorHandler.GetHistoryByName) } // WebSocket 路由（探针连接） @@ -186,6 +199,13 @@ func setupApi(app *orz.App, components *AppComponents) { // 告警记录查询 adminApi.GET("/alert-records", components.AlertHandler.ListAlertRecords) + // 服务监控配置 + adminApi.GET("/monitors", components.MonitorHandler.List) + adminApi.POST("/monitors", components.MonitorHandler.Create) + adminApi.GET("/monitors/:id", components.MonitorHandler.Get) + adminApi.PUT("/monitors/:id", components.MonitorHandler.Update) + adminApi.DELETE("/monitors/:id", components.MonitorHandler.Delete) + // 用户管理 adminApi.GET("/users", components.UserHandler.Paging) adminApi.POST("/users", components.UserHandler.Create) @@ -219,6 +239,8 @@ func autoMigrate(database *gorm.DB) error { &models.AlertConfig{}, &models.AlertRecord{}, &models.MonitorMetric{}, + &models.MonitorTask{}, + &models.MonitorStats{}, ) } @@ -364,6 +386,35 @@ func startMetricsMonitoring(ctx context.Context, components *AppComponents, logg } } +// startMonitorStatsCalculation 启动监控统计计算任务 +func startMonitorStatsCalculation(ctx context.Context, components *AppComponents, logger *zap.Logger) { + logger.Info("启动监控统计计算任务") + + ticker := time.NewTicker(5 * time.Minute) // 每5分钟计算一次统计数据 + defer ticker.Stop() + + // 首次启动时立即计算一次 + if err := components.MonitorService.CalculateMonitorStats(ctx); err != nil { + logger.Error("计算监控统计数据失败", zap.Error(err)) + } else { + logger.Info("监控统计数据计算完成") + } + + for { + select { + case <-ctx.Done(): + logger.Info("监控统计计算任务已停止") + return + case <-ticker.C: + if err := components.MonitorService.CalculateMonitorStats(ctx); err != nil { + logger.Error("计算监控统计数据失败", zap.Error(err)) + } else { + logger.Debug("监控统计数据计算完成") + } + } + } +} + // JWTAuthMiddleware JWT 认证中间件 func JWTAuthMiddleware(accountHandler *handler.AccountHandler) echo.MiddlewareFunc { return func(next echo.HandlerFunc) echo.HandlerFunc { @@ -396,3 +447,5 @@ func JWTAuthMiddleware(accountHandler *handler.AccountHandler) echo.MiddlewareFu } } } + +// APIKeyAuthMiddleware 使用 API Key 进行认证 diff --git a/internal/config/config.go b/internal/config/config.go index 13b8dbe..8031301 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,7 +2,8 @@ package config // AppConfig 应用配置 type AppConfig struct { - JWT JWTConfig `mapstructure:"JWT"` + JWT JWTConfig `mapstructure:"JWT"` + MonitorInterval int `mapstructure:"MonitorInterval"` } // JWTConfig JWT配置 diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index 2ba2457..1685180 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -25,15 +25,17 @@ import ( type AgentHandler struct { logger *zap.Logger agentService *service.AgentService + monitorSvc *service.MonitorService wsManager *ws.Manager upgrader websocket.Upgrader } -func NewAgentHandler(logger *zap.Logger, agentService *service.AgentService, wsManager *ws.Manager) *AgentHandler { +func NewAgentHandler(logger *zap.Logger, agentService *service.AgentService, monitorService *service.MonitorService, wsManager *ws.Manager) *AgentHandler { h := &AgentHandler{ logger: logger, agentService: agentService, + monitorSvc: monitorService, wsManager: wsManager, } @@ -210,7 +212,7 @@ func (h *AgentHandler) Paging(c echo.Context) error { pr := orz.GetPageRequest(c, "last_seen_at", "hostname") - builder := orz.NewPageBuilder(h.agentService.Repository). + builder := orz.NewPageBuilder(h.agentService.AgentRepo). PageRequest(pr). Contains("hostname", hostname). Contains("ip", ip) @@ -285,20 +287,18 @@ func (h *AgentHandler) GetMetrics(c echo.Context) error { } switch rangeParam { + case "1m": + start = end - 1*60*1000 + case "5m": + start = end - 5*60*1000 + case "15m": + start = end - 15*60*1000 + case "30m": + start = end - 30*60*1000 case "1h": - start = end - 1*60*60*1000 - case "6h": - start = end - 6*60*60*1000 - case "12h": - start = end - 12*60*60*1000 - case "24h", "1d": - start = end - 24*60*60*1000 - case "3d": - start = end - 3*24*60*60*1000 - case "7d", "1w": - start = end - 7*24*60*60*1000 + start = end - 60*60*1000 default: - return orz.NewError(400, "无效的时间范围，支持: 1h, 6h, 12h, 24h, 3d, 7d, 30d, 90d") + return orz.NewError(400, "无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h") } // 服务端自动计算最优聚合间隔 @@ -582,22 +582,18 @@ func (h *AgentHandler) GetMonitorMetrics(c echo.Context) error { } switch rangeParam { + case "1m": + start = end - 1*60*1000 + case "5m": + start = end - 5*60*1000 + case "15m": + start = end - 15*60*1000 + case "30m": + start = end - 30*60*1000 case "1h": - start = end - 1*60*60*1000 - case "6h": - start = end - 6*60*60*1000 - case "12h": - start = end - 12*60*60*1000 - case "24h", "1d": - start = end - 24*60*60*1000 - case "3d": - start = end - 3*24*60*60*1000 - case "7d", "1w": - start = end - 7*24*60*60*1000 - case "30d", "1M": - start = end - 30*24*60*60*1000 + start = end - 60*60*1000 default: - return orz.NewError(400, "无效的时间范围，支持: 1h, 6h, 12h, 24h, 3d, 7d, 30d") + return orz.NewError(400, "无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h") } metrics, err := h.agentService.GetMonitorMetrics(ctx, agentID, monitorName, start, end) diff --git a/internal/handler/api_key_handler.go b/internal/handler/api_key_handler.go index d75ca63..3888894 100644 --- a/internal/handler/api_key_handler.go +++ b/internal/handler/api_key_handler.go @@ -35,7 +35,7 @@ func (r ApiKeyHandler) Paging(c echo.Context) error { pr := orz.GetPageRequest(c, "created_at", "name") - builder := orz.NewPageBuilder(r.apiKeyService.Repository). + builder := orz.NewPageBuilder(r.apiKeyService.ApiKeyRepo). PageRequest(pr). Contains("name", name) diff --git a/internal/handler/monitor_handler.go b/internal/handler/monitor_handler.go new file mode 100644 index 0000000..baf62a7 --- /dev/null +++ b/internal/handler/monitor_handler.go @@ -0,0 +1,162 @@ +package handler + +import ( + "github.com/dushixiang/pika/internal/service" + "github.com/go-orz/orz" + "github.com/labstack/echo/v4" + "go.uber.org/zap" +) + +type MonitorHandler struct { + logger *zap.Logger + monitorService *service.MonitorService +} + +func NewMonitorHandler(logger *zap.Logger, monitorService *service.MonitorService) *MonitorHandler { + return &MonitorHandler{ + logger: logger, + monitorService: monitorService, + } +} + +func (h *MonitorHandler) List(c echo.Context) error { + keyword := c.QueryParam("keyword") + name := c.QueryParam("name") + target := c.QueryParam("target") + typeParam := c.QueryParam("type") + enabled := c.QueryParam("enabled") + + pr := orz.GetPageRequest(c, "-updated_at", "name") + + builder := orz.NewPageBuilder(h.monitorService.MonitorRepo). + PageRequest(pr) + + // 如果有 keyword，则搜索 name（也可以搜索 target） + if keyword != "" { + // keyword 可以匹配 name 或 target + builder.Contains("name", keyword) + // 如果需要同时搜索 target，可以手动构建 OR 查询 + // 这里简化处理，只搜索 name + } else { + // 否则使用独立的 name 和 target 搜索 + builder.Contains("name", name). + Contains("target", target) + } + + // 处理类型筛选 + if typeParam != "" { + builder.Equal("type", typeParam) + } + + // 处理启用状态筛选 + if enabled == "true" { + builder.Equal("enabled", "1") + } else if enabled == "false" { + builder.Equal("enabled", "0") + } + + ctx := c.Request().Context() + page, err := builder.Execute(ctx) + if err != nil { + return err + } + return orz.Ok(c, page) +} + +func (h *MonitorHandler) Create(c echo.Context) error { + var req service.MonitorTaskRequest + if err := c.Bind(&req); err != nil { + return orz.NewError(400, "请求参数错误") + } + + ctx := c.Request().Context() + item, err := h.monitorService.CreateMonitor(ctx, &req) + if err != nil { + return err + } + + return orz.Ok(c, item) +} + +func (h *MonitorHandler) Get(c echo.Context) error { + id := c.Param("id") + + ctx := c.Request().Context() + item, err := h.monitorService.FindById(ctx, id) + if err != nil { + return err + } + + return orz.Ok(c, item) +} + +func (h *MonitorHandler) Update(c echo.Context) error { + id := c.Param("id") + + var req service.MonitorTaskRequest + if err := c.Bind(&req); err != nil { + return orz.NewError(400, "请求参数错误") + } + + ctx := c.Request().Context() + item, err := h.monitorService.UpdateMonitor(ctx, id, &req) + if err != nil { + return err + } + + return orz.Ok(c, item) +} + +func (h *MonitorHandler) Delete(c echo.Context) error { + id := c.Param("id") + + ctx := c.Request().Context() + if err := h.monitorService.DeleteMonitor(ctx, id); err != nil { + return err + } + + return nil +} + +// GetAllStats 获取所有监控统计数据 +func (h *MonitorHandler) GetAllStats(c echo.Context) error { + ctx := c.Request().Context() + stats, err := h.monitorService.GetAllMonitorStats(ctx) + if err != nil { + return err + } + + return orz.Ok(c, stats) +} + +// GetStatsByName 获取指定监控任务的统计数据（所有探针） +func (h *MonitorHandler) GetStatsByName(c echo.Context) error { + name := c.Param("name") + + ctx := c.Request().Context() + stats, err := h.monitorService.GetMonitorStatsByName(ctx, name) + if err != nil { + return err + } + + return orz.Ok(c, stats) +} + +// GetHistoryByName 获取指定监控任务的历史响应时间数据 +func (h *MonitorHandler) GetHistoryByName(c echo.Context) error { + name := c.Param("name") + timeRange := c.QueryParam("range") + + // 默认时间范围为 5 分钟 + if timeRange == "" { + timeRange = "5m" + } + + ctx := c.Request().Context() + history, err := h.monitorService.GetMonitorHistory(ctx, name, timeRange) + if err != nil { + return err + } + + return orz.Ok(c, history) +} diff --git a/internal/handler/user_handler.go b/internal/handler/user_handler.go index 0d336ca..dd48f91 100644 --- a/internal/handler/user_handler.go +++ b/internal/handler/user_handler.go @@ -32,7 +32,7 @@ func (r UserHandler) Paging(c echo.Context) error { pr := orz.GetPageRequest(c, "created_at", "username") - builder := orz.NewPageBuilder(r.userService.Repository). + builder := orz.NewPageBuilder(r.userService.UserRepo). PageRequest(pr). Contains("username", username) diff --git a/internal/models/metric.go b/internal/models/metric.go index 173b701..fe921b5 100644 --- a/internal/models/metric.go +++ b/internal/models/metric.go @@ -179,18 +179,20 @@ func (DockerMetric) TableName() string { // MonitorMetric 监控指标 type MonitorMetric struct { - ID uint `gorm:"primaryKey;autoIncrement" json:"id"` - AgentID string `gorm:"index:idx_agent_name_time;index:idx_agent_time" json:"agentId"` // 探针ID - Name string `gorm:"index:idx_agent_name_time;index" json:"name"` // 监控项名称 - Type string `json:"type"` // 监控类型: http, tcp - Target string `json:"target"` // 监控目标 - Status string `json:"status"` // 状态: up, down - StatusCode int `json:"statusCode"` // HTTP状态码 - ResponseTime int64 `json:"responseTime"` // 响应时间(毫秒) - Error string `json:"error"` // 错误信息 - Message string `json:"message"` // 附加信息 - ContentMatch bool `json:"contentMatch"` // 内容匹配结果 - Timestamp int64 `gorm:"index:idx_agent_name_time;index:idx_agent_time;index:idx_time" json:"timestamp"` // 时间戳（毫秒） + ID uint `gorm:"primaryKey;autoIncrement" json:"id"` + AgentID string `gorm:"index:idx_agent_name_time;index:idx_agent_time" json:"agentId"` // 探针ID + Name string `gorm:"index:idx_agent_name_time;index" json:"name"` // 监控项名称 + Type string `json:"type"` // 监控类型: http, tcp + Target string `json:"target"` // 监控目标 + Status string `json:"status"` // 状态: up, down + StatusCode int `json:"statusCode"` // HTTP状态码 + ResponseTime int64 `json:"responseTime"` // 响应时间(毫秒) + Error string `json:"error"` // 错误信息 + Message string `json:"message"` // 附加信息 + ContentMatch bool `json:"contentMatch"` // 内容匹配结果 + CertExpiryTime int64 `json:"certExpiryTime"` // 证书过期时间(毫秒时间戳), 0表示无证书 + CertDaysLeft int `json:"certDaysLeft"` // 证书剩余天数 + Timestamp int64 `gorm:"index:idx_agent_name_time;index:idx_agent_time;index:idx_time" json:"timestamp"` // 时间戳（毫秒） } func (MonitorMetric) TableName() string { diff --git a/internal/models/monitor_task.go b/internal/models/monitor_task.go new file mode 100644 index 0000000..77e4e1c --- /dev/null +++ b/internal/models/monitor_task.go @@ -0,0 +1,26 @@ +package models + +import ( + "github.com/dushixiang/pika/internal/protocol" + "gorm.io/datatypes" +) + +// MonitorTask 描述一个服务监控任务 +type MonitorTask struct { + ID string `gorm:"primaryKey" json:"id"` // 任务 ID + Name string `gorm:"uniqueIndex" json:"name"` // 任务名称 + Type string `gorm:"index" json:"type"` // 监控类型 http/tcp + Target string `json:"target"` // 目标地址 + Description string `json:"description"` // 描述信息 + Enabled bool `json:"enabled"` // 是否启用 + Interval int `json:"interval"` // 检测频率（秒），默认 60 + AgentIds datatypes.JSONSlice[string] `json:"agentIds"` // 指定的探针 ID 列表（JSON 数组） + HTTPConfig datatypes.JSONType[protocol.HTTPMonitorConfig] `json:"httpConfig"` // HTTP 监控配置 + TCPConfig datatypes.JSONType[protocol.TCPMonitorConfig] `json:"tcpConfig"` // TCP 监控配置 + CreatedAt int64 `gorm:"autoCreateTime:milli" json:"createdAt"` // 创建时间 + UpdatedAt int64 `gorm:"autoUpdateTime:milli" json:"updatedAt"` // 更新时间 +} + +func (MonitorTask) TableName() string { + return "monitor_tasks" +} diff --git a/internal/protocol/message.go b/internal/protocol/message.go index 69b42a8..9bd6ddb 100644 --- a/internal/protocol/message.go +++ b/internal/protocol/message.go @@ -48,7 +48,8 @@ const ( MessageTypeCommand MessageType = "command" MessageTypeCommandResp MessageType = "command_response" // 指标消息 - MessageTypeMetrics MessageType = "metrics" + MessageTypeMetrics MessageType = "metrics" + MessageTypeMonitorConfig MessageType = "monitor_config" ) type MetricType string @@ -276,4 +277,7 @@ type MonitorData struct { CheckedAt int64 `json:"checkedAt"` // 检测时间(毫秒时间戳) Message string `json:"message,omitempty"` // 附加信息 ContentMatch bool `json:"contentMatch,omitempty"` // 内容匹配结果 + // TLS 证书信息（仅用于 HTTPS） + CertExpiryTime int64 `json:"certExpiryTime,omitempty"` // 证书过期时间(毫秒时间戳) + CertDaysLeft int `json:"certDaysLeft,omitempty"` // 证书剩余天数 } diff --git a/internal/protocol/monitor.go b/internal/protocol/monitor.go new file mode 100644 index 0000000..7e1150e --- /dev/null +++ b/internal/protocol/monitor.go @@ -0,0 +1,31 @@ +package protocol + +// MonitorConfigPayload 监控配置 payload +type MonitorConfigPayload struct { + Interval int `json:"interval"` + Items []MonitorItem `json:"items"` +} + +// MonitorItem 监控项配置 +type MonitorItem struct { + Name string `json:"name"` + Type string `json:"type"` + Target string `json:"target"` + HTTPConfig *HTTPMonitorConfig `json:"httpConfig,omitempty"` + TCPConfig *TCPMonitorConfig `json:"tcpConfig,omitempty"` +} + +// HTTPMonitorConfig HTTP 监控配置 +type HTTPMonitorConfig struct { + Method string `json:"method"` + ExpectedStatusCode int `json:"expectedStatusCode"` + ExpectedContent string `json:"expectedContent,omitempty"` + Timeout int `json:"timeout"` + Headers map[string]string `json:"headers,omitempty"` + Body string `json:"body,omitempty"` +} + +// TCPMonitorConfig TCP 监控配置 +type TCPMonitorConfig struct { + Timeout int `json:"timeout"` +} diff --git a/internal/providers.go b/internal/providers.go deleted file mode 100644 index fcc08fb..0000000 --- a/internal/providers.go +++ /dev/null @@ -1,80 +0,0 @@ -package internal - -import ( - "github.com/dushixiang/pika/internal/config" - "github.com/dushixiang/pika/internal/handler" - "github.com/dushixiang/pika/internal/repo" - "github.com/dushixiang/pika/internal/service" - "github.com/dushixiang/pika/internal/websocket" - "go.uber.org/zap" - "gorm.io/gorm" -) - -// provideAccountService 提供AccountService -func provideAccountService(logger *zap.Logger, userService *service.UserService, cfg *config.AppConfig) *service.AccountService { - return service.NewAccountService(logger, userService, cfg.JWT.Secret, cfg.JWT.ExpiresHours) -} - -// provideApiKeyRepo 提供ApiKeyRepo -func provideApiKeyRepo(db *gorm.DB) *repo.ApiKeyRepo { - return repo.NewApiKeyRepo(db) -} - -// provideApiKeyService 提供ApiKeyService -func provideApiKeyService(logger *zap.Logger, apiKeyRepo *repo.ApiKeyRepo) *service.ApiKeyService { - return service.NewApiKeyService(logger, apiKeyRepo) -} - -// provideApiKeyHandler 提供ApiKeyHandler -func provideApiKeyHandler(logger *zap.Logger, apiKeyService *service.ApiKeyService) *handler.ApiKeyHandler { - return handler.NewApiKeyHandler(logger, apiKeyService) -} - -// provideAgentService 提供AgentService -func provideAgentService(logger *zap.Logger, agentRepo *repo.AgentRepo, metricRepo *repo.MetricRepo, apiKeyService *service.ApiKeyService) *service.AgentService { - return service.NewAgentService(logger, agentRepo, metricRepo, apiKeyService) -} - -// provideAgentHandler 提供AgentHandler -func provideAgentHandler(logger *zap.Logger, agentService *service.AgentService, wsManager *websocket.Manager, cfg *config.AppConfig) *handler.AgentHandler { - return handler.NewAgentHandler( - logger, - agentService, - wsManager, - ) -} - -// provideAlertRepo 提供AlertRepo -func provideAlertRepo(db *gorm.DB) *repo.AlertRepo { - return repo.NewAlertRepo(db) -} - -// provideNotifier 提供Notifier -func provideNotifier(logger *zap.Logger) *service.Notifier { - return service.NewNotifier(logger) -} - -// providePropertyRepo 提供PropertyRepo -func providePropertyRepo(db *gorm.DB) *repo.PropertyRepo { - return repo.NewPropertyRepo(db) -} - -// providePropertyService 提供PropertyService -func providePropertyService(propertyRepo *repo.PropertyRepo, logger *zap.Logger) *service.PropertyService { - return service.NewPropertyService(propertyRepo, logger) -} - -// providePropertyHandler 提供PropertyHandler -func providePropertyHandler(logger *zap.Logger, propertyService *service.PropertyService, notifier *service.Notifier) *handler.PropertyHandler { - return handler.NewPropertyHandler(logger, propertyService, notifier) -} - -// provideAlertService 提供AlertService -func provideAlertService(alertRepo *repo.AlertRepo, agentRepo *repo.AgentRepo, propertyService *service.PropertyService, notifier *service.Notifier, logger *zap.Logger) *service.AlertService { - return service.NewAlertService(alertRepo, agentRepo, propertyService, notifier, logger) -} - -// provideAlertHandler 提供AlertHandler -func provideAlertHandler(logger *zap.Logger, alertService *service.AlertService) *handler.AlertHandler { - return handler.NewAlertHandler(logger, alertService) -} diff --git a/internal/repo/agent_repo.go b/internal/repo/agent_repo.go index cff9e1a..a7a4473 100644 --- a/internal/repo/agent_repo.go +++ b/internal/repo/agent_repo.go @@ -140,3 +140,15 @@ func (r *AgentRepo) GetStatistics(ctx context.Context) (total int64, online int6 return total, online, err } + +// ListByIDs 根据ID列表获取探针 +func (r *AgentRepo) ListByIDs(ctx context.Context, ids []string) ([]models.Agent, error) { + var agents []models.Agent + if len(ids) == 0 { + return agents, nil + } + err := r.db.WithContext(ctx). + Where("id IN ?", ids). + Find(&agents).Error + return agents, err +} diff --git a/internal/repo/metric_repo.go b/internal/repo/metric_repo.go index 4926038..818ed9d 100644 --- a/internal/repo/metric_repo.go +++ b/internal/repo/metric_repo.go @@ -57,28 +57,27 @@ func (r *MetricRepo) SaveTemperatureMetric(ctx context.Context, metric *models.T return r.db.WithContext(ctx).Create(metric).Error } -// SaveDockerMetric 保存Docker容器指标 -func (r *MetricRepo) SaveDockerMetric(ctx context.Context, metric *models.DockerMetric) error { - return r.db.WithContext(ctx).Create(metric).Error +// ReplaceDockerMetrics 用最新一次采集结果替换 Docker 容器指标 +func (r *MetricRepo) ReplaceDockerMetrics(ctx context.Context, agentID string, metrics []models.DockerMetric) error { + return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + // 清空该探针历史数据，只保留最新一次采集 + if err := tx.Where("agent_id = ?", agentID).Delete(&models.DockerMetric{}).Error; err != nil { + return err + } + if len(metrics) == 0 { + return nil + } + return tx.Create(&metrics).Error + }) } // GetLatestDockerMetrics 获取最新的Docker容器指标列表 func (r *MetricRepo) GetLatestDockerMetrics(ctx context.Context, agentID string) ([]models.DockerMetric, error) { var metrics []models.DockerMetric - // 获取每个容器的最新记录 err := r.db.WithContext(ctx). - Raw(` - SELECT d1.* FROM docker_metrics d1 - INNER JOIN ( - SELECT container_id, MAX(timestamp) as max_timestamp - FROM docker_metrics - WHERE agent_id = ? - GROUP BY container_id - ) d2 ON d1.container_id = d2.container_id AND d1.timestamp = d2.max_timestamp - WHERE d1.agent_id = ? - ORDER BY d1.name - `, agentID, agentID). - Scan(&metrics).Error + Where("agent_id = ?", agentID). + Order("name"). + Find(&metrics).Error return metrics, err } @@ -190,22 +189,22 @@ func (r *MetricRepo) DeleteOldMetrics(ctx context.Context, beforeTimestamp int64 return nil } -// AggregatedCPUMetric CPU聚合指标 +// AggregatedCPUMetric CPU聚合指标（使用最大值） type AggregatedCPUMetric struct { Timestamp int64 `json:"timestamp"` - AvgUsage float64 `json:"avgUsage"` + MaxUsage float64 `json:"maxUsage"` LogicalCores int `json:"logicalCores"` } -// GetCPUMetrics 获取聚合后的CPU指标（始终返回聚合数据） -// interval: 聚合间隔，单位秒（如：60表示1分钟，3600表示1小时） +// GetCPUMetrics 获取聚合后的CPU指标（取最大值） +// interval: 聚合间隔，单位秒（如：60表示1分钟） func (r *MetricRepo) GetCPUMetrics(ctx context.Context, agentID string, start, end int64, interval int) ([]AggregatedCPUMetric, error) { var metrics []AggregatedCPUMetric query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - AVG(usage_percent) as avg_usage, + MAX(usage_percent) as max_usage, MAX(logical_cores) as logical_cores FROM cpu_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? @@ -221,21 +220,21 @@ func (r *MetricRepo) GetCPUMetrics(ctx context.Context, agentID string, start, e return metrics, err } -// AggregatedMemoryMetric 内存聚合指标 +// AggregatedMemoryMetric 内存聚合指标（使用最大值） type AggregatedMemoryMetric struct { Timestamp int64 `json:"timestamp"` - AvgUsage float64 `json:"avgUsage"` + MaxUsage float64 `json:"maxUsage"` Total uint64 `json:"total"` } -// GetMemoryMetrics 获取聚合后的内存指标（始终返回聚合数据） +// GetMemoryMetrics 获取聚合后的内存指标（取最大值） func (r *MetricRepo) GetMemoryMetrics(ctx context.Context, agentID string, start, end int64, interval int) ([]AggregatedMemoryMetric, error) { var metrics []AggregatedMemoryMetric query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - AVG(usage_percent) as avg_usage, + MAX(usage_percent) as max_usage, MAX(total) as total FROM memory_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? @@ -251,11 +250,11 @@ func (r *MetricRepo) GetMemoryMetrics(ctx context.Context, agentID string, start return metrics, err } -// AggregatedDiskMetric 磁盘聚合指标 +// AggregatedDiskMetric 磁盘聚合指标（使用最大值） type AggregatedDiskMetric struct { Timestamp int64 `json:"timestamp"` MountPoint string `json:"mountPoint"` - AvgUsage float64 `json:"avgUsage"` + MaxUsage float64 `json:"maxUsage"` Total uint64 `json:"total"` } @@ -267,7 +266,7 @@ func (r *MetricRepo) GetDiskMetrics(ctx context.Context, agentID string, start, SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, mount_point, - AVG(usage_percent) as avg_usage, + MAX(usage_percent) as max_usage, MAX(total) as total FROM disk_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? @@ -283,11 +282,11 @@ func (r *MetricRepo) GetDiskMetrics(ctx context.Context, agentID string, start, return metrics, err } -// AggregatedNetworkMetric 网络聚合指标 +// AggregatedNetworkMetric 网络聚合指标（使用最大值） type AggregatedNetworkMetric struct { Timestamp int64 `json:"timestamp"` - AvgSentRate float64 `json:"avgSentRate"` - AvgRecvRate float64 `json:"avgRecvRate"` + MaxSentRate float64 `json:"maxSentRate"` + MaxRecvRate float64 `json:"maxRecvRate"` } // GetNetworkMetrics 获取聚合后的网络指标（合并所有网卡接口） @@ -297,8 +296,8 @@ func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, star query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - AVG(bytes_sent_rate) as avg_sent_rate, - AVG(bytes_recv_rate) as avg_recv_rate + MAX(bytes_sent_rate) as max_sent_rate, + MAX(bytes_recv_rate) as max_recv_rate FROM network_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? GROUP BY 1 @@ -313,12 +312,12 @@ func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, star return metrics, err } -// AggregatedLoadMetric 负载聚合指标 +// AggregatedLoadMetric 负载聚合指标（使用最大值） type AggregatedLoadMetric struct { Timestamp int64 `json:"timestamp"` - AvgLoad1 float64 `json:"avgLoad1"` - AvgLoad5 float64 `json:"avgLoad5"` - AvgLoad15 float64 `json:"avgLoad15"` + MaxLoad1 float64 `json:"maxLoad1"` + MaxLoad5 float64 `json:"maxLoad5"` + MaxLoad15 float64 `json:"maxLoad15"` } // GetLoadMetrics 获取聚合后的负载指标（始终返回聚合数据） @@ -328,9 +327,9 @@ func (r *MetricRepo) GetLoadMetrics(ctx context.Context, agentID string, start, query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - AVG(load1) as avg_load1, - AVG(load5) as avg_load5, - AVG(load15) as avg_load15 + MAX(load1) as max_load1, + MAX(load5) as max_load5, + MAX(load15) as max_load15 FROM load_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? GROUP BY 1 @@ -494,12 +493,67 @@ func (r *MetricRepo) GetMonitorMetricsByName(ctx context.Context, agentID, monit return metrics, err } +// AggregatedMonitorMetric 聚合的监控指标 +type AggregatedMonitorMetric struct { + Timestamp int64 `json:"timestamp"` + AgentID string `json:"agentId"` + AvgResponse float64 `json:"avgResponse"` // 平均响应时间 + MaxResponse int64 `json:"maxResponse"` // 最大响应时间 + MinResponse int64 `json:"minResponse"` // 最小响应时间 + SuccessCount int64 `json:"successCount"` // 成功次数 + TotalCount int64 `json:"totalCount"` // 总次数 + SuccessRate float64 `json:"successRate"` // 成功率 + LastStatus string `json:"lastStatus"` // 最后状态 + LastErrorMsg string `json:"lastErrorMsg"` // 最后错误信息 +} + +// GetAggregatedMonitorMetrics 获取聚合后的监控指标（按探针和时间间隔聚合） +func (r *MetricRepo) GetAggregatedMonitorMetrics(ctx context.Context, monitorName string, start, end int64, interval int) ([]AggregatedMonitorMetric, error) { + var metrics []AggregatedMonitorMetric + + query := ` + WITH ranked_metrics AS ( + SELECT + agent_id, + CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as time_bucket, + response_time, + status, + error, + timestamp, + ROW_NUMBER() OVER (PARTITION BY agent_id, CAST(FLOOR(timestamp / ?) * ? AS BIGINT) ORDER BY timestamp DESC) as rn + FROM monitor_metrics + WHERE name = ? AND timestamp >= ? AND timestamp <= ? + ) + SELECT + time_bucket as timestamp, + agent_id, + AVG(response_time) as avg_response, + MAX(response_time) as max_response, + MIN(response_time) as min_response, + SUM(CASE WHEN status = 'up' THEN 1 ELSE 0 END) as success_count, + COUNT(*) as total_count, + CAST(SUM(CASE WHEN status = 'up' THEN 1 ELSE 0 END) AS REAL) / CAST(COUNT(*) AS REAL) * 100 as success_rate, + MAX(CASE WHEN rn = 1 THEN status END) as last_status, + MAX(CASE WHEN rn = 1 THEN error END) as last_error_msg + FROM ranked_metrics + GROUP BY time_bucket, agent_id + ORDER BY timestamp ASC, agent_id + ` + + intervalMs := int64(interval * 1000) + err := r.db.WithContext(ctx). + Raw(query, intervalMs, intervalMs, intervalMs, intervalMs, monitorName, start, end). + Scan(&metrics).Error + + return metrics, err +} + // AggregatedDiskIOMetric 磁盘IO聚合指标 type AggregatedDiskIOMetric struct { Timestamp int64 `json:"timestamp"` Device string `json:"device"` - AvgReadRate float64 `json:"avgReadRate"` // 平均读取速率(字节/秒) - AvgWriteRate float64 `json:"avgWriteRate"` // 平均写入速率(字节/秒) + MaxReadRate float64 `json:"maxReadRate"` // 最大读取速率(字节/秒) + MaxWriteRate float64 `json:"maxWriteRate"` // 最大写入速率(字节/秒) TotalReadBytes uint64 `json:"totalReadBytes"` // 总读取字节数 TotalWriteBytes uint64 `json:"totalWriteBytes"` // 总写入字节数 } @@ -513,8 +567,8 @@ func (r *MetricRepo) GetDiskIOMetrics(ctx context.Context, agentID string, start SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, device, - AVG(read_bytes_rate) as avg_read_rate, - AVG(write_bytes_rate) as avg_write_rate, + MAX(read_bytes_rate) as max_read_rate, + MAX(write_bytes_rate) as max_write_rate, CASE WHEN MAX(read_bytes) >= MIN(read_bytes) THEN MAX(read_bytes) - MIN(read_bytes) ELSE MAX(read_bytes) @@ -537,13 +591,13 @@ func (r *MetricRepo) GetDiskIOMetrics(ctx context.Context, agentID string, start return metrics, err } -// AggregatedGPUMetric GPU聚合指标 +// AggregatedGPUMetric GPU聚合指标（使用最大值） type AggregatedGPUMetric struct { Timestamp int64 `json:"timestamp"` - AvgUtilization float64 `json:"avgUtilization"` - AvgMemoryUsed float64 `json:"avgMemoryUsed"` - AvgTemperature float64 `json:"avgTemperature"` - AvgPowerDraw float64 `json:"avgPowerDraw"` + MaxUtilization float64 `json:"maxUtilization"` + MaxMemoryUsed uint64 `json:"maxMemoryUsed"` + MaxTemperature float64 `json:"maxTemperature"` + MaxPowerDraw float64 `json:"maxPowerDraw"` } // GetGPUMetrics 获取聚合后的GPU指标 @@ -553,10 +607,10 @@ func (r *MetricRepo) GetGPUMetrics(ctx context.Context, agentID string, start, e query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - AVG(utilization) as avg_utilization, - AVG(memory_used) as avg_memory_used, - AVG(temperature) as avg_temperature, - AVG(power_draw) as avg_power_draw + MAX(utilization) as max_utilization, + MAX(memory_used) as max_memory_used, + MAX(temperature) as max_temperature, + MAX(power_draw) as max_power_draw FROM gpu_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? GROUP BY 1 @@ -571,12 +625,12 @@ func (r *MetricRepo) GetGPUMetrics(ctx context.Context, agentID string, start, e return metrics, err } -// AggregatedTemperatureMetric 温度聚合指标 +// AggregatedTemperatureMetric 温度聚合指标（使用最大值） type AggregatedTemperatureMetric struct { Timestamp int64 `json:"timestamp"` SensorKey string `json:"sensorKey"` SensorLabel string `json:"sensorLabel"` - AvgTemperature float64 `json:"avgTemperature"` + MaxTemperature float64 `json:"maxTemperature"` } // GetTemperatureMetrics 获取聚合后的温度指标 @@ -588,7 +642,7 @@ func (r *MetricRepo) GetTemperatureMetrics(ctx context.Context, agentID string, CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, sensor_key, sensor_label, - AVG(temperature) as avg_temperature + MAX(temperature) as max_temperature FROM temperature_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? GROUP BY 1, sensor_key, sensor_label diff --git a/internal/repo/monitor_repo.go b/internal/repo/monitor_repo.go new file mode 100644 index 0000000..ad39d01 --- /dev/null +++ b/internal/repo/monitor_repo.go @@ -0,0 +1,31 @@ +package repo + +import ( + "context" + + "github.com/dushixiang/pika/internal/models" + "github.com/go-orz/orz" + "gorm.io/gorm" +) + +type MonitorRepo struct { + orz.Repository[models.MonitorTask, string] +} + +func NewMonitorRepo(db *gorm.DB) *MonitorRepo { + return &MonitorRepo{ + Repository: orz.NewRepository[models.MonitorTask, string](db), + } +} + +func (r *MonitorRepo) FindByEnabledAndAgentId(ctx context.Context, agentId string) ([]models.MonitorTask, error) { + var tasks []models.MonitorTask + if err := r.GetDB(ctx). + Model(&models.MonitorTask{}). + Where(`enabled = ? and (agent_id = ? or agent_id = "")`, true, agentId). + Find(&tasks).Error; err != nil { + return nil, err + } + + return tasks, nil +} diff --git a/internal/scheduler/monitor_scheduler.go b/internal/scheduler/monitor_scheduler.go new file mode 100644 index 0000000..116800e --- /dev/null +++ b/internal/scheduler/monitor_scheduler.go @@ -0,0 +1,283 @@ +package scheduler + +import ( + "context" + "sync" + "time" + + "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/service" + "go.uber.org/zap" +) + +// MonitorTask 调度任务 +type MonitorTask struct { + ID string + Monitor models.MonitorTask + NextRunTime time.Time + Interval time.Duration + Running bool +} + +// MonitorScheduler 监控任务调度器 +type MonitorScheduler struct { + mu sync.RWMutex + tasks map[string]*MonitorTask // taskID -> MonitorTask + monitorService *service.MonitorService + logger *zap.Logger + ctx context.Context + cancel context.CancelFunc + workerCount int + taskChan chan *MonitorTask + reloadInterval time.Duration + tickInterval time.Duration +} + +// NewMonitorScheduler 创建监控任务调度器 +func NewMonitorScheduler(monitorService *service.MonitorService, logger *zap.Logger, workerCount int) *MonitorScheduler { + if workerCount <= 0 { + workerCount = 5 // 默认 5 个 worker + } + + return &MonitorScheduler{ + tasks: make(map[string]*MonitorTask), + monitorService: monitorService, + logger: logger, + workerCount: workerCount, + taskChan: make(chan *MonitorTask, 100), + reloadInterval: 10 * time.Second, // 每 10 秒重新加载任务列表 + tickInterval: 1 * time.Second, // 每秒检查一次 + } +} + +// Start 启动调度器 +func (s *MonitorScheduler) Start(ctx context.Context) { + s.ctx, s.cancel = context.WithCancel(ctx) + + s.logger.Info("启动监控任务调度器", + zap.Int("workerCount", s.workerCount)) + + // 启动 worker pool + for i := 0; i < s.workerCount; i++ { + go s.worker(i) + } + + // 首次加载任务 + s.reloadTasks() + + // 启动调度循环 + go s.scheduleLoop() + + // 启动重载循环 + go s.reloadLoop() +} + +// Stop 停止调度器 +func (s *MonitorScheduler) Stop() { + if s.cancel != nil { + s.cancel() + } + close(s.taskChan) + s.logger.Info("监控任务调度器已停止") +} + +// scheduleLoop 调度循环 +func (s *MonitorScheduler) scheduleLoop() { + ticker := time.NewTicker(s.tickInterval) + defer ticker.Stop() + + for { + select { + case <-s.ctx.Done(): + return + case now := <-ticker.C: + s.checkAndSchedule(now) + } + } +} + +// reloadLoop 重载循环 +func (s *MonitorScheduler) reloadLoop() { + ticker := time.NewTicker(s.reloadInterval) + defer ticker.Stop() + + for { + select { + case <-s.ctx.Done(): + return + case <-ticker.C: + s.reloadTasks() + } + } +} + +// checkAndSchedule 检查并调度应该执行的任务 +func (s *MonitorScheduler) checkAndSchedule(now time.Time) { + s.mu.RLock() + tasksToRun := make([]*MonitorTask, 0) + + for _, task := range s.tasks { + // 如果任务未在运行且到了执行时间 + if !task.Running && !now.Before(task.NextRunTime) { + tasksToRun = append(tasksToRun, task) + } + } + s.mu.RUnlock() + + // 提交任务到 worker pool + for _, task := range tasksToRun { + s.mu.Lock() + task.Running = true + s.mu.Unlock() + + select { + case s.taskChan <- task: + // 任务已提交 + default: + // worker pool 已满，重置运行状态 + s.mu.Lock() + task.Running = false + s.mu.Unlock() + s.logger.Warn("worker pool 已满，跳过任务", + zap.String("taskID", task.ID), + zap.String("taskName", task.Monitor.Name)) + } + } +} + +// worker worker 协程，执行任务 +func (s *MonitorScheduler) worker(id int) { + s.logger.Debug("启动 worker", zap.Int("workerID", id)) + + for { + select { + case <-s.ctx.Done(): + return + case task, ok := <-s.taskChan: + if !ok { + return + } + s.executeTask(task) + } + } +} + +// executeTask 执行任务 +func (s *MonitorScheduler) executeTask(task *MonitorTask) { + defer func() { + // 更新下次执行时间和状态 + s.mu.Lock() + task.NextRunTime = time.Now().Add(task.Interval) + task.Running = false + s.mu.Unlock() + }() + + s.logger.Debug("执行监控任务", + zap.String("taskID", task.ID), + zap.String("taskName", task.Monitor.Name), + zap.Duration("interval", task.Interval)) + + // 发送监控任务到探针 + if err := s.monitorService.SendMonitorTaskToAgents(s.ctx, task.Monitor, task.Monitor.AgentIds); err != nil { + s.logger.Error("发送监控任务失败", + zap.String("taskID", task.ID), + zap.String("taskName", task.Monitor.Name), + zap.Error(err)) + } +} + +// reloadTasks 重新加载任务列表 +func (s *MonitorScheduler) reloadTasks() { + // 获取所有启用的监控任务 + var monitors []models.MonitorTask + if err := s.monitorService.MonitorRepo.GetDB(s.ctx). + Where("enabled = ?", true). + Find(&monitors).Error; err != nil { + s.logger.Error("加载监控任务失败", zap.Error(err)) + return + } + + s.mu.Lock() + defer s.mu.Unlock() + + // 标记当前存在的任务 + existingTasks := make(map[string]bool) + for _, monitor := range monitors { + existingTasks[monitor.ID] = true + + if task, exists := s.tasks[monitor.ID]; exists { + // 任务已存在，检查是否需要更新 + if task.Monitor.Interval != monitor.Interval || + task.Monitor.Name != monitor.Name || + task.Monitor.Type != monitor.Type || + task.Monitor.Target != monitor.Target { + // 任务参数已变化，更新任务 + s.logger.Info("更新监控任务", + zap.String("taskID", monitor.ID), + zap.String("taskName", monitor.Name)) + task.Monitor = monitor + task.Interval = time.Duration(monitor.Interval) * time.Second + if task.Interval <= 0 { + task.Interval = 60 * time.Second + } + } + } else { + // 新任务，添加到调度器 + interval := time.Duration(monitor.Interval) * time.Second + if interval <= 0 { + interval = 60 * time.Second + } + + s.logger.Info("添加监控任务", + zap.String("taskID", monitor.ID), + zap.String("taskName", monitor.Name), + zap.Duration("interval", interval)) + + s.tasks[monitor.ID] = &MonitorTask{ + ID: monitor.ID, + Monitor: monitor, + NextRunTime: time.Now(), // 立即执行一次 + Interval: interval, + Running: false, + } + } + } + + // 删除已不存在或已禁用的任务 + for taskID := range s.tasks { + if !existingTasks[taskID] { + s.logger.Info("删除监控任务", zap.String("taskID", taskID)) + delete(s.tasks, taskID) + } + } +} + +// GetTaskCount 获取任务数量 +func (s *MonitorScheduler) GetTaskCount() int { + s.mu.RLock() + defer s.mu.RUnlock() + return len(s.tasks) +} + +// GetTaskStatus 获取任务状态 +func (s *MonitorScheduler) GetTaskStatus() map[string]interface{} { + s.mu.RLock() + defer s.mu.RUnlock() + + tasks := make([]map[string]interface{}, 0, len(s.tasks)) + for _, task := range s.tasks { + tasks = append(tasks, map[string]interface{}{ + "id": task.ID, + "name": task.Monitor.Name, + "interval": task.Interval.Seconds(), + "nextRunTime": task.NextRunTime.Format(time.RFC3339), + "running": task.Running, + }) + } + + return map[string]interface{}{ + "totalTasks": len(s.tasks), + "workerCount": s.workerCount, + "tasks": tasks, + } +} diff --git a/internal/service/account_service.go b/internal/service/account_service.go index a28942a..75b9b98 100644 --- a/internal/service/account_service.go +++ b/internal/service/account_service.go @@ -4,13 +4,17 @@ import ( "context" "time" + "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/models" "github.com/go-errors/errors" "github.com/golang-jwt/jwt/v5" "go.uber.org/zap" ) -func NewAccountService(logger *zap.Logger, userService *UserService, jwtSecret string, tokenExpireHours int) *AccountService { +func NewAccountService(logger *zap.Logger, userService *UserService, appConfig *config.AppConfig) *AccountService { + jwtSecret := appConfig.JWT.Secret + tokenExpireHours := appConfig.JWT.ExpiresHours + if jwtSecret == "" { logger.Fatal("JWT secret cannot be empty") } diff --git a/internal/service/agent_service.go b/internal/service/agent_service.go index 13f604d..9086bae 100644 --- a/internal/service/agent_service.go +++ b/internal/service/agent_service.go @@ -10,23 +10,22 @@ import ( "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/internal/repo" "go.uber.org/zap" + "gorm.io/gorm" ) type AgentService struct { logger *zap.Logger - agentRepo *repo.AgentRepo + AgentRepo *repo.AgentRepo metricRepo *repo.MetricRepo apiKeyService *ApiKeyService - Repository *repo.AgentRepo } -func NewAgentService(logger *zap.Logger, agentRepo *repo.AgentRepo, metricRepo *repo.MetricRepo, apiKeyService *ApiKeyService) *AgentService { +func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService) *AgentService { return &AgentService{ logger: logger, - agentRepo: agentRepo, - metricRepo: metricRepo, + AgentRepo: repo.NewAgentRepo(db), + metricRepo: repo.NewMetricRepo(db), apiKeyService: apiKeyService, - Repository: agentRepo, } } @@ -48,7 +47,7 @@ func (s *AgentService) RegisterAgent(ctx context.Context, ip string, info *proto // 使用探针的持久化 ID 来识别同一个探针 // 这样即使主机名或 IP 变化，也能正确识别 - existingAgent, err := s.agentRepo.FindById(ctx, info.ID) + existingAgent, err := s.AgentRepo.FindById(ctx, info.ID) if err == nil { // 更新现有探针信息（允许主机名、IP、名称等变化） now := time.Now().UnixMilli() @@ -61,7 +60,7 @@ func (s *AgentService) RegisterAgent(ctx context.Context, ip string, info *proto existingAgent.LastSeenAt = now existingAgent.UpdatedAt = now - if err := s.agentRepo.UpdateById(ctx, &existingAgent); err != nil { + if err := s.AgentRepo.UpdateById(ctx, &existingAgent); err != nil { return nil, err } s.logger.Info("agent re-registered", @@ -89,7 +88,7 @@ func (s *AgentService) RegisterAgent(ctx context.Context, ip string, info *proto UpdatedAt: now, } - if err := s.agentRepo.Create(ctx, agent); err != nil { + if err := s.AgentRepo.Create(ctx, agent); err != nil { return nil, err } @@ -104,12 +103,12 @@ 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) } // GetAgent 获取探针信息 func (s *AgentService) GetAgent(ctx context.Context, agentID string) (*models.Agent, error) { - agent, err := s.agentRepo.FindById(ctx, agentID) + agent, err := s.AgentRepo.FindById(ctx, agentID) if err != nil { return nil, err } @@ -118,12 +117,12 @@ func (s *AgentService) GetAgent(ctx context.Context, agentID string) (*models.Ag // ListAgents 列出所有探针 func (s *AgentService) ListAgents(ctx context.Context) ([]models.Agent, error) { - return s.agentRepo.FindAll(ctx) + return s.AgentRepo.FindAll(ctx) } // ListOnlineAgents 列出所有在线探针 func (s *AgentService) ListOnlineAgents(ctx context.Context) ([]models.Agent, error) { - return s.agentRepo.FindOnlineAgents(ctx) + return s.AgentRepo.FindOnlineAgents(ctx) } // HandleMetricData 处理指标数据 @@ -342,9 +341,9 @@ func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, met if err := json.Unmarshal(data, &dockerDataList); err != nil { return err } - // 保存每个容器的数据 + metrics := make([]models.DockerMetric, 0, len(dockerDataList)) for _, dockerData := range dockerDataList { - metric := &models.DockerMetric{ + metrics = append(metrics, models.DockerMetric{ AgentID: agentID, ContainerID: dockerData.ContainerID, Name: dockerData.Name, @@ -361,13 +360,13 @@ func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, met BlockOutput: dockerData.BlockOutput, Pids: dockerData.Pids, Timestamp: now, - } - if err := s.metricRepo.SaveDockerMetric(ctx, metric); err != nil { - s.logger.Error("failed to save docker metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("containerID", dockerData.ContainerID)) - } + }) + } + if err := s.metricRepo.ReplaceDockerMetrics(ctx, agentID, metrics); err != nil { + s.logger.Error("failed to replace docker metrics", + zap.Error(err), + zap.String("agentID", agentID)) + return err } return nil @@ -380,17 +379,19 @@ func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, met // 保存每个监控项的数据 for _, monitorData := range monitorDataList { metric := &models.MonitorMetric{ - AgentID: agentID, - Name: monitorData.Name, - Type: monitorData.Type, - Target: monitorData.Target, - Status: monitorData.Status, - StatusCode: monitorData.StatusCode, - ResponseTime: monitorData.ResponseTime, - Error: monitorData.Error, - Message: monitorData.Message, - ContentMatch: monitorData.ContentMatch, - Timestamp: monitorData.CheckedAt, // 使用检测时间 + AgentID: agentID, + Name: monitorData.Name, + Type: monitorData.Type, + Target: monitorData.Target, + Status: monitorData.Status, + StatusCode: monitorData.StatusCode, + ResponseTime: monitorData.ResponseTime, + Error: monitorData.Error, + Message: monitorData.Message, + ContentMatch: monitorData.ContentMatch, + CertExpiryTime: monitorData.CertExpiryTime, + CertDaysLeft: monitorData.CertDaysLeft, + Timestamp: monitorData.CheckedAt, // 使用检测时间 } if err := s.metricRepo.SaveMonitorMetric(ctx, metric); err != nil { s.logger.Error("failed to save monitor metric", @@ -408,25 +409,21 @@ func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, met } // CalculateInterval 根据时间范围计算合适的聚合间隔（秒） -// 目标是返回大约200-500个数据点 +// 目标是返回尽量平滑的曲线，同时控制数据点数量 func CalculateInterval(start, end int64) int { duration := (end - start) / 1000 // 转换为秒 switch { - case duration <= 3600: // 1小时内: 10秒聚合 - return 10 - case duration <= 21600: // 6小时内: 1分钟聚合 + case duration <= 60: // 1分钟内: 2秒 + return 2 + case duration <= 300: // 5分钟内: 5秒 + return 5 + case duration <= 900: // 15分钟内: 15秒 + return 15 + case duration <= 1800: // 30分钟内: 30秒 + return 30 + default: // 1小时内: 60秒 return 60 - case duration <= 86400: // 1天内: 5分钟聚合 - return 300 - case duration <= 259200: // 3天内: 15分钟聚合 - return 900 - case duration <= 604800: // 7天内: 30分钟聚合 - return 1800 - case duration <= 2592000: // 30天内: 2小时聚合 - return 7200 - default: // 30天以上: 6小时聚合 - return 21600 } } @@ -518,6 +515,9 @@ func (s *AgentService) GetLatestMetrics(ctx context.Context, agentID string) (*L // 获取最新Docker容器信息 if docker, err := s.metricRepo.GetLatestDockerMetrics(ctx, agentID); err == nil && len(docker) > 0 { + for i := range docker { + docker[i].Image = "" + } result.Docker = docker } @@ -536,7 +536,7 @@ func (s *AgentService) GetLatestMetrics(ctx context.Context, agentID string) (*L // StartCleanupTask 启动数据清理任务 func (s *AgentService) StartCleanupTask(ctx context.Context) { - ticker := time.NewTicker(1 * time.Hour) + ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() s.logger.Info("cleanup task started") @@ -554,8 +554,8 @@ func (s *AgentService) StartCleanupTask(ctx context.Context) { // cleanupOldMetrics 清理旧数据 func (s *AgentService) cleanupOldMetrics(ctx context.Context) { - // 删除7天前的数据 - before := time.Now().AddDate(0, 0, -7).UnixMilli() + // 删除1小时前的数据 + before := time.Now().Add(-1 * time.Hour).UnixMilli() s.logger.Info("starting to clean old metrics", zap.Int64("beforeTimestamp", before)) @@ -663,12 +663,12 @@ func (s *AgentService) SaveAuditResult(ctx context.Context, agentID string, resu } // 保存到数据库 - return s.agentRepo.SaveAuditResult(ctx, auditRecord) + return s.AgentRepo.SaveAuditResult(ctx, auditRecord) } // GetAuditResult 获取最新的审计结果 func (s *AgentService) GetAuditResult(ctx context.Context, agentID string) (*protocol.VPSAuditResult, error) { - record, err := s.agentRepo.GetLatestAuditResult(ctx, agentID) + record, err := s.AgentRepo.GetLatestAuditResult(ctx, agentID) if err != nil { return nil, err } @@ -683,7 +683,7 @@ func (s *AgentService) GetAuditResult(ctx context.Context, agentID string) (*pro // ListAuditResults 获取审计结果列表 func (s *AgentService) ListAuditResults(ctx context.Context, agentID string) ([]map[string]interface{}, error) { - records, err := s.agentRepo.ListAuditResults(ctx, agentID) + records, err := s.AgentRepo.ListAuditResults(ctx, agentID) if err != nil { return nil, err } @@ -731,17 +731,17 @@ func (s *AgentService) ListAuditResults(ctx context.Context, agentID string) ([] // UpdateAgentName 更新探针名称 func (s *AgentService) UpdateAgentName(ctx context.Context, agentID string, name string) error { - return s.agentRepo.UpdateName(ctx, agentID, name) + return s.AgentRepo.UpdateName(ctx, agentID, name) } // UpdateAgentInfo 更新探针信息（名称、平台、位置、到期时间） func (s *AgentService) UpdateAgentInfo(ctx context.Context, agentID string, updates map[string]interface{}) error { - return s.agentRepo.UpdateInfo(ctx, agentID, updates) + return s.AgentRepo.UpdateInfo(ctx, agentID, updates) } // GetStatistics 获取探针统计数据 func (s *AgentService) GetStatistics(ctx context.Context) (map[string]interface{}, error) { - total, online, err := s.agentRepo.GetStatistics(ctx) + total, online, err := s.AgentRepo.GetStatistics(ctx) if err != nil { return nil, err } diff --git a/internal/service/alert_service.go b/internal/service/alert_service.go index 2d19bb6..d946672 100644 --- a/internal/service/alert_service.go +++ b/internal/service/alert_service.go @@ -10,6 +10,7 @@ import ( "github.com/dushixiang/pika/internal/repo" "github.com/google/uuid" "go.uber.org/zap" + "gorm.io/gorm" ) // AlertService 告警服务 @@ -25,10 +26,10 @@ type AlertService struct { mu sync.RWMutex } -func NewAlertService(alertRepo *repo.AlertRepo, agentRepo *repo.AgentRepo, propertyService *PropertyService, notifier *Notifier, logger *zap.Logger) *AlertService { +func NewAlertService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, notifier *Notifier) *AlertService { return &AlertService{ - alertRepo: alertRepo, - agentRepo: agentRepo, + alertRepo: repo.NewAlertRepo(db), + agentRepo: repo.NewAgentRepo(db), propertyService: propertyService, notifier: notifier, logger: logger, diff --git a/internal/service/api_key_service.go b/internal/service/api_key_service.go index c3d4827..fa39b5b 100644 --- a/internal/service/api_key_service.go +++ b/internal/service/api_key_service.go @@ -11,19 +11,18 @@ import ( "github.com/dushixiang/pika/internal/repo" "github.com/google/uuid" "go.uber.org/zap" + "gorm.io/gorm" ) type ApiKeyService struct { logger *zap.Logger - apiKeyRepo *repo.ApiKeyRepo - Repository *repo.ApiKeyRepo + ApiKeyRepo *repo.ApiKeyRepo } -func NewApiKeyService(logger *zap.Logger, apiKeyRepo *repo.ApiKeyRepo) *ApiKeyService { +func NewApiKeyService(logger *zap.Logger, db *gorm.DB) *ApiKeyService { return &ApiKeyService{ logger: logger, - apiKeyRepo: apiKeyRepo, - Repository: apiKeyRepo, + ApiKeyRepo: repo.NewApiKeyRepo(db), } } @@ -46,7 +45,7 @@ func (s *ApiKeyService) GenerateApiKey(ctx context.Context, name, userID string) UpdatedAt: now, } - if err := s.apiKeyRepo.Create(ctx, apiKey); err != nil { + if err := s.ApiKeyRepo.Create(ctx, apiKey); err != nil { return nil, err } @@ -64,7 +63,7 @@ func (s *ApiKeyService) ValidateApiKey(ctx context.Context, key string) (*models return nil, errors.New("api key is required") } - apiKey, err := s.apiKeyRepo.FindEnabledByKey(ctx, key) + apiKey, err := s.ApiKeyRepo.FindEnabledByKey(ctx, key) if err != nil { s.logger.Warn("invalid api key", zap.String("key", key)) return nil, errors.New("invalid api key") @@ -75,7 +74,7 @@ func (s *ApiKeyService) ValidateApiKey(ctx context.Context, key string) (*models // GetApiKey 获取API密钥信息 func (s *ApiKeyService) GetApiKey(ctx context.Context, id string) (*models.ApiKey, error) { - apiKey, err := s.apiKeyRepo.FindById(ctx, id) + apiKey, err := s.ApiKeyRepo.FindById(ctx, id) if err != nil { return nil, err } @@ -84,17 +83,17 @@ func (s *ApiKeyService) GetApiKey(ctx context.Context, id string) (*models.ApiKe // ListApiKeys 列出所有API密钥 func (s *ApiKeyService) ListApiKeys(ctx context.Context, page, pageSize int) ([]models.ApiKey, int64, error) { - return s.apiKeyRepo.ListAll(ctx, page, pageSize) + return s.ApiKeyRepo.ListAll(ctx, page, pageSize) } // ListUserApiKeys 列出用户的API密钥 func (s *ApiKeyService) ListUserApiKeys(ctx context.Context, userID string, page, pageSize int) ([]models.ApiKey, int64, error) { - return s.apiKeyRepo.ListByUser(ctx, userID, page, pageSize) + return s.ApiKeyRepo.ListByUser(ctx, userID, page, pageSize) } // UpdateApiKeyName 更新API密钥名称 func (s *ApiKeyService) UpdateApiKeyName(ctx context.Context, id, name string) error { - if err := s.apiKeyRepo.UpdateName(ctx, id, name); err != nil { + if err := s.ApiKeyRepo.UpdateName(ctx, id, name); err != nil { return err } @@ -107,7 +106,7 @@ func (s *ApiKeyService) UpdateApiKeyName(ctx context.Context, id, name string) e // EnableApiKey 启用API密钥 func (s *ApiKeyService) EnableApiKey(ctx context.Context, id string) error { - if err := s.apiKeyRepo.UpdateEnabled(ctx, id, true); err != nil { + if err := s.ApiKeyRepo.UpdateEnabled(ctx, id, true); err != nil { return err } @@ -117,7 +116,7 @@ func (s *ApiKeyService) EnableApiKey(ctx context.Context, id string) error { // DisableApiKey 禁用API密钥 func (s *ApiKeyService) DisableApiKey(ctx context.Context, id string) error { - if err := s.apiKeyRepo.UpdateEnabled(ctx, id, false); err != nil { + if err := s.ApiKeyRepo.UpdateEnabled(ctx, id, false); err != nil { return err } @@ -127,7 +126,7 @@ func (s *ApiKeyService) DisableApiKey(ctx context.Context, id string) error { // DeleteApiKey 删除API密钥 func (s *ApiKeyService) DeleteApiKey(ctx context.Context, id string) error { - if err := s.apiKeyRepo.DeleteById(ctx, id); err != nil { + if err := s.ApiKeyRepo.DeleteById(ctx, id); err != nil { return err } diff --git a/internal/service/monitor_service.go b/internal/service/monitor_service.go new file mode 100644 index 0000000..f4e3c07 --- /dev/null +++ b/internal/service/monitor_service.go @@ -0,0 +1,477 @@ +package service + +import ( + "context" + "encoding/json" + "strings" + "time" + + "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/protocol" + "github.com/dushixiang/pika/internal/repo" + ws "github.com/dushixiang/pika/internal/websocket" + "github.com/go-orz/orz" + "github.com/google/uuid" + "go.uber.org/zap" + "gorm.io/datatypes" + "gorm.io/gorm" +) + +type MonitorService struct { + logger *zap.Logger + *repo.MonitorRepo + *orz.Service + agentRepo *repo.AgentRepo + metricRepo *repo.MetricRepo + monitorStatsRepo *repo.MonitorStatsRepo + wsManager *ws.Manager +} + +func NewMonitorService(logger *zap.Logger, db *gorm.DB, wsManager *ws.Manager) *MonitorService { + return &MonitorService{ + logger: logger, + MonitorRepo: repo.NewMonitorRepo(db), + agentRepo: repo.NewAgentRepo(db), + metricRepo: repo.NewMetricRepo(db), + monitorStatsRepo: repo.NewMonitorStatsRepo(db), + wsManager: wsManager, + } +} + +type MonitorTaskRequest struct { + Name string `json:"name"` + Type string `json:"type"` + Target string `json:"target"` + Description string `json:"description"` + Enabled bool `json:"enabled,omitempty"` + Interval int `json:"interval"` // 检测频率（秒） + HTTPConfig protocol.HTTPMonitorConfig `json:"httpConfig,omitempty"` + TCPConfig protocol.TCPMonitorConfig `json:"tcpConfig,omitempty"` + AgentIds []string `json:"agentIds,omitempty"` +} + +func (s *MonitorService) CreateMonitor(ctx context.Context, req *MonitorTaskRequest) (*models.MonitorTask, error) { + // 设置默认检测频率 + interval := req.Interval + if interval <= 0 { + interval = 60 // 默认 60 秒 + } + + task := &models.MonitorTask{ + ID: uuid.NewString(), + Name: strings.TrimSpace(req.Name), + Type: req.Type, + Target: strings.TrimSpace(req.Target), + Description: req.Description, + Enabled: req.Enabled, + Interval: interval, + AgentIds: datatypes.JSONSlice[string](req.AgentIds), + HTTPConfig: datatypes.NewJSONType(req.HTTPConfig), + TCPConfig: datatypes.NewJSONType(req.TCPConfig), + CreatedAt: 0, + UpdatedAt: 0, + } + + if err := s.MonitorRepo.Create(ctx, task); err != nil { + return nil, err + } + + return task, nil +} + +func (s *MonitorService) UpdateMonitor(ctx context.Context, id string, req *MonitorTaskRequest) (*models.MonitorTask, error) { + task, err := s.MonitorRepo.FindById(ctx, id) + if err != nil { + return nil, err + } + + task.Name = strings.TrimSpace(req.Name) + task.Type = req.Type + task.Target = strings.TrimSpace(req.Target) + task.Description = req.Description + + // 更新检测频率 + interval := req.Interval + if interval <= 0 { + interval = 60 // 默认 60 秒 + } + task.Interval = interval + + task.AgentIds = req.AgentIds + task.HTTPConfig = datatypes.NewJSONType(req.HTTPConfig) + task.TCPConfig = datatypes.NewJSONType(req.TCPConfig) + + if err := s.MonitorRepo.UpdateById(ctx, &task); err != nil { + return nil, err + } + + return &task, nil +} + +func (s *MonitorService) DeleteMonitor(ctx context.Context, id string) error { + return s.Transaction(ctx, func(ctx context.Context) error { + if err := s.MonitorRepo.DeleteById(ctx, id); err != nil { + return err + } + return nil + }) +} + +// BroadcastMonitorConfig 向所有在线探针广播监控配置 +func (s *MonitorService) BroadcastMonitorConfig(ctx context.Context) error { + // 获取所有启用的监控任务 + var monitors []models.MonitorTask + if err := s.MonitorRepo.GetDB(ctx). + Where("enabled = ?", true). + Find(&monitors).Error; err != nil { + return err + } + + // 如果没有启用的监控任务，直接返回（不需要发送任何配置） + if len(monitors) == 0 { + s.logger.Debug("没有启用的监控任务，跳过配置推送") + return nil + } + + // 获取所有在线探针 + agents, err := s.agentRepo.FindOnlineAgents(ctx) + if err != nil { + s.logger.Error("获取在线探针失败", zap.Error(err)) + return err + } + + // 按探针分组构建监控配置 + agentMonitors := make(map[string][]models.MonitorTask) + + for _, monitor := range monitors { + // 如果没有指定探针，则发送给所有探针 + if len(monitor.AgentIds) == 0 { + for _, agent := range agents { + agentMonitors[agent.ID] = append(agentMonitors[agent.ID], monitor) + } + } else { + // 只发送给指定的探针（只发送给在线的探针） + for _, agentID := range monitor.AgentIds { + // 检查该探针是否在线 + isOnline := false + for _, agent := range agents { + if agent.ID == agentID { + isOnline = true + break + } + } + if isOnline { + agentMonitors[agentID] = append(agentMonitors[agentID], monitor) + } + } + } + } + + // 向每个有监控任务的探针发送对应的监控配置 + for agentID, tasks := range agentMonitors { + items := make([]protocol.MonitorItem, 0, len(tasks)) + for _, task := range tasks { + item := protocol.MonitorItem{ + Name: task.Name, + Type: task.Type, + Target: task.Target, + } + + if task.Type == "http" || task.Type == "https" { + var httpConfig protocol.HTTPMonitorConfig + if err := task.HTTPConfig.Scan(&httpConfig); err == nil { + item.HTTPConfig = &httpConfig + } + } else if task.Type == "tcp" { + var tcpConfig protocol.TCPMonitorConfig + if err := task.TCPConfig.Scan(&tcpConfig); err == nil { + item.TCPConfig = &tcpConfig + } + } + + items = append(items, item) + } + + // 构建监控配置 payload + // Interval 字段不再使用（探针收到后立即检测一次），但保留字段兼容性 + payload := protocol.MonitorConfigPayload{ + Interval: 0, + Items: items, + } + + // 发送监控配置到指定探针 + if err := s.sendMonitorConfigToAgent(agentID, payload); err != nil { + s.logger.Error("发送监控配置失败", + zap.String("agentID", agentID), + zap.Error(err)) + } else { + s.logger.Debug("发送监控配置成功", + zap.String("agentID", agentID), + zap.Int("taskCount", len(items))) + } + } + + return nil +} + +// sendMonitorConfigToAgent 向指定探针发送监控配置（内部方法） +func (s *MonitorService) sendMonitorConfigToAgent(agentID string, payload protocol.MonitorConfigPayload) error { + payloadData, err := json.Marshal(payload) + if err != nil { + return err + } + + msg := protocol.Message{ + Type: protocol.MessageTypeMonitorConfig, + Data: payloadData, + } + + msgData, err := json.Marshal(msg) + if err != nil { + return err + } + + return s.wsManager.SendToClient(agentID, msgData) +} + +// SendMonitorTaskToAgents 向指定探针发送单个监控任务（公开方法） +func (s *MonitorService) SendMonitorTaskToAgents(ctx context.Context, monitor models.MonitorTask, agentIDs []string) error { + // 获取所有在线探针 + agents, err := s.agentRepo.FindOnlineAgents(ctx) + if err != nil { + return err + } + + // 确定要发送的探针列表 + var targetAgents []models.Agent + if len(agentIDs) == 0 { + // 发送给所有在线探针 + targetAgents = agents + } else { + // 只发送给指定的在线探针 + for _, agent := range agents { + for _, agentID := range agentIDs { + if agent.ID == agentID { + targetAgents = append(targetAgents, agent) + break + } + } + } + } + + if len(targetAgents) == 0 { + return nil + } + + // 构建监控项 + item := protocol.MonitorItem{ + Name: monitor.Name, + Type: monitor.Type, + Target: monitor.Target, + } + + if monitor.Type == "http" || monitor.Type == "https" { + var httpConfig protocol.HTTPMonitorConfig + if err := monitor.HTTPConfig.Scan(&httpConfig); err == nil { + item.HTTPConfig = &httpConfig + } + } else if monitor.Type == "tcp" { + var tcpConfig protocol.TCPMonitorConfig + if err := monitor.TCPConfig.Scan(&tcpConfig); err == nil { + item.TCPConfig = &tcpConfig + } + } + + // 构建 payload + payload := protocol.MonitorConfigPayload{ + Interval: 0, + Items: []protocol.MonitorItem{item}, + } + + // 向每个目标探针发送 + for _, agent := range targetAgents { + if err := s.sendMonitorConfigToAgent(agent.ID, payload); err != nil { + s.logger.Error("发送监控配置失败", + zap.String("taskID", monitor.ID), + zap.String("taskName", monitor.Name), + zap.String("agentID", agent.ID), + zap.Error(err)) + } + } + + return nil +} + +// CalculateMonitorStats 计算监控统计数据 +func (s *MonitorService) CalculateMonitorStats(ctx context.Context) error { + now := time.Now() + + // 获取所有启用的监控任务 + var monitors []models.MonitorTask + if err := s.MonitorRepo.GetDB(ctx). + Where("enabled = ?", true). + Find(&monitors).Error; err != nil { + return err + } + + // 获取所有在线探针 + agents, err := s.agentRepo.FindOnlineAgents(ctx) + if err != nil { + return err + } + + // 为每个监控任务的每个探针计算统计数据 + for _, monitor := range monitors { + var targetAgents []models.Agent + if len(monitor.AgentIds) == 0 { + targetAgents = agents + } else { + for _, agent := range agents { + for _, agentID := range monitor.AgentIds { + if agent.ID == agentID { + targetAgents = append(targetAgents, agent) + break + } + } + } + } + + for _, agent := range targetAgents { + stats, err := s.calculateStatsForAgentMonitor(ctx, agent.ID, monitor.Name, monitor.Type, monitor.Target, now) + if err != nil { + s.logger.Error("计算监控统计失败", + zap.String("agentID", agent.ID), + zap.String("monitorName", monitor.Name), + zap.Error(err)) + continue + } + + if err := s.monitorStatsRepo.UpsertStats(ctx, stats); err != nil { + s.logger.Error("保存监控统计失败", + zap.String("agentID", agent.ID), + zap.String("monitorName", monitor.Name), + zap.Error(err)) + } + } + } + + return nil +} + +// calculateStatsForAgentMonitor 计算单个探针单个监控任务的统计数据 +func (s *MonitorService) calculateStatsForAgentMonitor(ctx context.Context, agentID, monitorName, monitorType, target string, now time.Time) (*models.MonitorStats, error) { + stats := &models.MonitorStats{ + AgentID: agentID, + MonitorName: monitorName, + MonitorType: monitorType, + Target: target, + } + + // 计算24小时数据 + start24h := now.Add(-24 * time.Hour).UnixMilli() + end := now.UnixMilli() + metrics24h, err := s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorName, start24h, end) + if err != nil { + return nil, err + } + + // 计算30天数据 + start30d := now.Add(-30 * 24 * time.Hour).UnixMilli() + metrics30d, err := s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorName, start30d, end) + if err != nil { + return nil, err + } + + // 计算24小时统计 + if len(metrics24h) > 0 { + var totalResponse int64 + var successCount int64 + lastMetric := metrics24h[len(metrics24h)-1] + + for _, metric := range metrics24h { + if metric.Status == "up" { + successCount++ + totalResponse += metric.ResponseTime + } + } + + stats.TotalChecks24h = int64(len(metrics24h)) + stats.SuccessChecks24h = successCount + if successCount > 0 { + stats.AvgResponse24h = totalResponse / successCount + } + if stats.TotalChecks24h > 0 { + stats.Uptime24h = float64(successCount) / float64(stats.TotalChecks24h) * 100 + } + + // 最后一次检测数据 + stats.CurrentResponse = lastMetric.ResponseTime + stats.LastCheckTime = lastMetric.Timestamp + stats.LastCheckStatus = lastMetric.Status + + // 从最新的检测结果中获取证书信息 + if lastMetric.CertExpiryTime > 0 { + stats.CertExpiryDate = lastMetric.CertExpiryTime + stats.CertExpiryDays = lastMetric.CertDaysLeft + } + } + + // 计算30天统计 + if len(metrics30d) > 0 { + var successCount int64 + for _, metric := range metrics30d { + if metric.Status == "up" { + successCount++ + } + } + + stats.TotalChecks30d = int64(len(metrics30d)) + stats.SuccessChecks30d = successCount + if stats.TotalChecks30d > 0 { + stats.Uptime30d = float64(successCount) / float64(stats.TotalChecks30d) * 100 + } + } + + return stats, nil +} + +// GetMonitorStatsByName 获取监控任务的统计数据（所有探针） +func (s *MonitorService) GetMonitorStatsByName(ctx context.Context, monitorName string) ([]models.MonitorStats, error) { + return s.monitorStatsRepo.ListByMonitorName(ctx, monitorName) +} + +// GetAllMonitorStats 获取所有监控统计数据 +func (s *MonitorService) GetAllMonitorStats(ctx context.Context) ([]models.MonitorStats, error) { + return s.monitorStatsRepo.ListAll(ctx) +} + +// GetMonitorHistory 获取监控任务的历史响应时间数据 +func (s *MonitorService) GetMonitorHistory(ctx context.Context, monitorName, timeRange string) ([]repo.AggregatedMonitorMetric, error) { + // 解析时间范围 + var duration time.Duration + var interval int // 聚合间隔（秒） + + switch timeRange { + case "5m": + duration = 5 * time.Minute + interval = 15 // 15秒聚合一次 + case "15m": + duration = 15 * time.Minute + interval = 30 // 30秒聚合一次 + case "30m": + duration = 30 * time.Minute + interval = 60 // 1分钟聚合一次 + case "1h": + duration = 1 * time.Hour + interval = 120 // 2分钟聚合一次 + default: + duration = 5 * time.Minute + interval = 15 + } + + now := time.Now() + end := now.UnixMilli() + start := now.Add(-duration).UnixMilli() + + return s.metricRepo.GetAggregatedMonitorMetrics(ctx, monitorName, start, end, interval) +} diff --git a/internal/service/property_service.go b/internal/service/property_service.go index 8ddac74..c06dba5 100644 --- a/internal/service/property_service.go +++ b/internal/service/property_service.go @@ -9,6 +9,7 @@ import ( "github.com/dushixiang/pika/internal/models" "github.com/dushixiang/pika/internal/repo" "go.uber.org/zap" + "gorm.io/gorm" ) const ( @@ -21,9 +22,9 @@ type PropertyService struct { logger *zap.Logger } -func NewPropertyService(repo *repo.PropertyRepo, logger *zap.Logger) *PropertyService { +func NewPropertyService(logger *zap.Logger, db *gorm.DB) *PropertyService { return &PropertyService{ - repo: repo, + repo: repo.NewPropertyRepo(db), logger: logger, } } diff --git a/internal/service/user_service.go b/internal/service/user_service.go index afe5d25..781992e 100644 --- a/internal/service/user_service.go +++ b/internal/service/user_service.go @@ -10,26 +10,25 @@ import ( "github.com/google/uuid" "go.uber.org/zap" "golang.org/x/crypto/bcrypt" + "gorm.io/gorm" ) type UserService struct { - logger *zap.Logger - userRepo *repo.UserRepo - Repository *repo.UserRepo + logger *zap.Logger + UserRepo *repo.UserRepo } -func NewUserService(logger *zap.Logger, userRepo *repo.UserRepo) *UserService { +func NewUserService(logger *zap.Logger, db *gorm.DB) *UserService { return &UserService{ - logger: logger, - userRepo: userRepo, - Repository: userRepo, + logger: logger, + UserRepo: repo.NewUserRepo(db), } } // CreateUser 创建用户 func (s *UserService) CreateUser(ctx context.Context, req *CreateUserRequest) (*models.User, error) { // 检查用户名是否已存在 - existingUser, err := s.userRepo.FindByUsername(ctx, req.Username) + existingUser, err := s.UserRepo.FindByUsername(ctx, req.Username) if err == nil && existingUser != nil { return nil, errors.New("用户名已存在") } @@ -55,7 +54,7 @@ func (s *UserService) CreateUser(ctx context.Context, req *CreateUserRequest) (* UpdatedAt: now, } - if err := s.userRepo.Create(ctx, user); err != nil { + if err := s.UserRepo.Create(ctx, user); err != nil { return nil, err } @@ -65,7 +64,7 @@ func (s *UserService) CreateUser(ctx context.Context, req *CreateUserRequest) (* // UpdateUser 更新用户 func (s *UserService) UpdateUser(ctx context.Context, userID string, req *UpdateUserRequest) (*models.User, error) { - user, err := s.userRepo.FindById(ctx, userID) + user, err := s.UserRepo.FindById(ctx, userID) if err != nil { return nil, errors.New("用户不存在") } @@ -75,7 +74,7 @@ func (s *UserService) UpdateUser(ctx context.Context, userID string, req *Update user.Nickname = req.Nickname } - if err := s.userRepo.UpdateById(ctx, &user); err != nil { + if err := s.UserRepo.UpdateById(ctx, &user); err != nil { return nil, err } @@ -85,7 +84,7 @@ func (s *UserService) UpdateUser(ctx context.Context, userID string, req *Update // DeleteUser 删除用户 func (s *UserService) DeleteUser(ctx context.Context, userID string) error { - if err := s.userRepo.DeleteById(ctx, userID); err != nil { + if err := s.UserRepo.DeleteById(ctx, userID); err != nil { return err } @@ -95,7 +94,7 @@ func (s *UserService) DeleteUser(ctx context.Context, userID string) error { // GetUser 获取用户信息 func (s *UserService) GetUser(ctx context.Context, userID string) (*models.User, error) { - user, err := s.userRepo.FindById(ctx, userID) + user, err := s.UserRepo.FindById(ctx, userID) if err != nil { return nil, errors.New("用户不存在") } @@ -104,18 +103,18 @@ func (s *UserService) GetUser(ctx context.Context, userID string) (*models.User, // GetUserByUsername 根据用户名获取用户 func (s *UserService) GetUserByUsername(ctx context.Context, username string) (*models.User, error) { - return s.userRepo.FindByUsername(ctx, username) + return s.UserRepo.FindByUsername(ctx, username) } // ListUsers 列出用户 func (s *UserService) ListUsers(ctx context.Context, page, pageSize int) ([]models.User, int64, error) { offset := (page - 1) * pageSize - return s.userRepo.ListUsers(ctx, offset, pageSize) + return s.UserRepo.ListUsers(ctx, offset, pageSize) } // ChangePassword 修改密码 func (s *UserService) ChangePassword(ctx context.Context, userID string, oldPassword, newPassword string) error { - user, err := s.userRepo.FindById(ctx, userID) + user, err := s.UserRepo.FindById(ctx, userID) if err != nil { return errors.New("用户不存在") } @@ -131,7 +130,7 @@ func (s *UserService) ChangePassword(ctx context.Context, userID string, oldPass return err } - if err := s.userRepo.UpdatePassword(ctx, userID, string(hashedPassword)); err != nil { + if err := s.UserRepo.UpdatePassword(ctx, userID, string(hashedPassword)); err != nil { return err } @@ -147,7 +146,7 @@ func (s *UserService) ResetPassword(ctx context.Context, userID, newPassword str return err } - if err := s.userRepo.UpdatePassword(ctx, userID, string(hashedPassword)); err != nil { + if err := s.UserRepo.UpdatePassword(ctx, userID, string(hashedPassword)); err != nil { return err } @@ -157,7 +156,7 @@ func (s *UserService) ResetPassword(ctx context.Context, userID, newPassword str // UpdateUserStatus 更新用户状态 func (s *UserService) UpdateUserStatus(ctx context.Context, userID string, status int) error { - if err := s.userRepo.UpdateStatus(ctx, userID, status); err != nil { + if err := s.UserRepo.UpdateStatus(ctx, userID, status); err != nil { return err } @@ -167,7 +166,7 @@ func (s *UserService) UpdateUserStatus(ctx context.Context, userID string, statu // ValidatePassword 验证密码 func (s *UserService) ValidatePassword(ctx context.Context, username, password string) (*models.User, error) { - user, err := s.userRepo.FindByUsername(ctx, username) + user, err := s.UserRepo.FindByUsername(ctx, username) if err != nil { return nil, errors.New("用户名或密码错误") } diff --git a/internal/wire.go b/internal/wire.go index fb22ed5..d836bb4 100644 --- a/internal/wire.go +++ b/internal/wire.go @@ -6,7 +6,6 @@ package internal import ( "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/handler" - "github.com/dushixiang/pika/internal/repo" "github.com/dushixiang/pika/internal/service" "github.com/dushixiang/pika/internal/websocket" "github.com/google/wire" @@ -17,35 +16,26 @@ import ( // InitializeApp 初始化应用 func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*AppComponents, error) { wire.Build( - // Repositories - repo.NewAgentRepo, - repo.NewMetricRepo, - repo.NewUserRepo, - provideApiKeyRepo, - provideAlertRepo, - providePropertyRepo, - - // Services - provideApiKeyService, - provideAgentService, + service.NewAccountService, + service.NewAgentService, service.NewUserService, - provideNotifier, - providePropertyService, - provideAlertService, - - // Providers for services with config - provideAccountService, - provideAgentHandler, - provideApiKeyHandler, - provideAlertHandler, - providePropertyHandler, + service.NewApiKeyService, + service.NewAlertService, + service.NewPropertyService, + service.NewMonitorService, + service.NewNotifier, // WebSocket Manager websocket.NewManager, // Handlers - handler.NewAccountHandler, + handler.NewAgentHandler, handler.NewUserHandler, + handler.NewAlertHandler, + handler.NewPropertyHandler, + handler.NewMonitorHandler, + handler.NewApiKeyHandler, + handler.NewAccountHandler, // App Components wire.Struct(new(AppComponents), "*"), @@ -61,9 +51,14 @@ type AppComponents struct { ApiKeyHandler *handler.ApiKeyHandler AlertHandler *handler.AlertHandler PropertyHandler *handler.PropertyHandler + MonitorHandler *handler.MonitorHandler + AgentService *service.AgentService UserService *service.UserService AlertService *service.AlertService PropertyService *service.PropertyService - WSManager *websocket.Manager + MonitorService *service.MonitorService + ApiKeyService *service.ApiKeyService + + WSManager *websocket.Manager } diff --git a/internal/wire_gen.go b/internal/wire_gen.go index a352812..76446f0 100644 --- a/internal/wire_gen.go +++ b/internal/wire_gen.go @@ -9,7 +9,6 @@ package internal import ( "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/handler" - "github.com/dushixiang/pika/internal/repo" "github.com/dushixiang/pika/internal/service" "github.com/dushixiang/pika/internal/websocket" "go.uber.org/zap" @@ -20,26 +19,22 @@ import ( // InitializeApp 初始化应用 func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*AppComponents, error) { - userRepo := repo.NewUserRepo(db) - userService := service.NewUserService(logger, userRepo) - accountService := provideAccountService(logger, userService, cfg) + userService := service.NewUserService(logger, db) + accountService := service.NewAccountService(logger, userService, cfg) accountHandler := handler.NewAccountHandler(accountService) - agentRepo := repo.NewAgentRepo(db) - metricRepo := repo.NewMetricRepo(db) - apiKeyRepo := provideApiKeyRepo(db) - apiKeyService := provideApiKeyService(logger, apiKeyRepo) - agentService := provideAgentService(logger, agentRepo, metricRepo, apiKeyService) + apiKeyService := service.NewApiKeyService(logger, db) + agentService := service.NewAgentService(logger, db, apiKeyService) manager := websocket.NewManager(logger) - agentHandler := provideAgentHandler(logger, agentService, manager, cfg) + monitorService := service.NewMonitorService(logger, db, manager) + agentHandler := handler.NewAgentHandler(logger, agentService, monitorService, manager) userHandler := handler.NewUserHandler(userService) - apiKeyHandler := provideApiKeyHandler(logger, apiKeyService) - alertRepo := provideAlertRepo(db) - propertyRepo := providePropertyRepo(db) - propertyService := providePropertyService(propertyRepo, logger) - notifier := provideNotifier(logger) - alertService := provideAlertService(alertRepo, agentRepo, propertyService, notifier, logger) - alertHandler := provideAlertHandler(logger, alertService) - propertyHandler := providePropertyHandler(logger, propertyService, notifier) + apiKeyHandler := handler.NewApiKeyHandler(logger, apiKeyService) + propertyService := service.NewPropertyService(logger, db) + notifier := service.NewNotifier(logger) + alertService := service.NewAlertService(logger, db, propertyService, notifier) + alertHandler := handler.NewAlertHandler(logger, alertService) + propertyHandler := handler.NewPropertyHandler(logger, propertyService, notifier) + monitorHandler := handler.NewMonitorHandler(logger, monitorService) appComponents := &AppComponents{ AccountHandler: accountHandler, AgentHandler: agentHandler, @@ -47,10 +42,13 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App ApiKeyHandler: apiKeyHandler, AlertHandler: alertHandler, PropertyHandler: propertyHandler, + MonitorHandler: monitorHandler, AgentService: agentService, UserService: userService, AlertService: alertService, PropertyService: propertyService, + MonitorService: monitorService, + ApiKeyService: apiKeyService, WSManager: manager, } return appComponents, nil @@ -66,9 +64,14 @@ type AppComponents struct { ApiKeyHandler *handler.ApiKeyHandler AlertHandler *handler.AlertHandler PropertyHandler *handler.PropertyHandler + MonitorHandler *handler.MonitorHandler + AgentService *service.AgentService UserService *service.UserService AlertService *service.AlertService PropertyService *service.PropertyService - WSManager *websocket.Manager + MonitorService *service.MonitorService + ApiKeyService *service.ApiKeyService + + WSManager *websocket.Manager } diff --git a/pkg/agent/collector/docker.go b/pkg/agent/collector/docker.go index c208ade..94eb0f5 100644 --- a/pkg/agent/collector/docker.go +++ b/pkg/agent/collector/docker.go @@ -101,40 +101,6 @@ type containerStats struct { Pids int } -// statsJSON Docker 容器统计信息的 JSON 结构 -type statsJSON struct { - CPUStats struct { - CPUUsage struct { - TotalUsage uint64 `json:"total_usage"` - } `json:"cpu_usage"` - SystemUsage uint64 `json:"system_cpu_usage"` - OnlineCPUs uint32 `json:"online_cpus"` - } `json:"cpu_stats"` - PreCPUStats struct { - CPUUsage struct { - TotalUsage uint64 `json:"total_usage"` - } `json:"cpu_usage"` - SystemUsage uint64 `json:"system_cpu_usage"` - } `json:"precpu_stats"` - MemoryStats struct { - Usage uint64 `json:"usage"` - Limit uint64 `json:"limit"` - } `json:"memory_stats"` - BlkioStats struct { - IoServiceBytesRecursive []struct { - Op string `json:"op"` - Value uint64 `json:"value"` - } `json:"io_service_bytes_recursive"` - } `json:"blkio_stats"` - Networks map[string]struct { - RxBytes uint64 `json:"rx_bytes"` - TxBytes uint64 `json:"tx_bytes"` - } `json:"networks"` - PidsStats struct { - Current uint64 `json:"current"` - } `json:"pids_stats"` -} - // getContainerStats 获取容器统计信息 func (d *DockerCollector) getContainerStats(ctx context.Context, containerID string) (*containerStats, error) { // 获取容器统计信息（不使用流式传输） @@ -150,7 +116,7 @@ func (d *DockerCollector) getContainerStats(ctx context.Context, containerID str return nil, err } - var stats statsJSON + var stats container.StatsResponse if err := json.Unmarshal(body, &stats); err != nil { return nil, err } @@ -188,7 +154,7 @@ func (d *DockerCollector) getContainerStats(ctx context.Context, containerID str } // calculateCPUPercent 计算 CPU 使用率 -func calculateCPUPercent(stats *statsJSON) float64 { +func calculateCPUPercent(stats *container.StatsResponse) float64 { cpuDelta := float64(stats.CPUStats.CPUUsage.TotalUsage - stats.PreCPUStats.CPUUsage.TotalUsage) systemDelta := float64(stats.CPUStats.SystemUsage - stats.PreCPUStats.SystemUsage) diff --git a/pkg/agent/collector/manager.go b/pkg/agent/collector/manager.go index 7db7055..f723cf7 100644 --- a/pkg/agent/collector/manager.go +++ b/pkg/agent/collector/manager.go @@ -40,7 +40,7 @@ func NewManager(cfg *config.Config) *Manager { temperatureCollector: NewTemperatureCollector(), gpuCollector: NewGPUCollector(), dockerCollector: NewDockerCollector(), - monitorCollector: NewMonitorCollector(cfg), + monitorCollector: NewMonitorCollector(), } } @@ -144,13 +144,8 @@ func (m *Manager) CollectAndSendDocker(conn WebSocketWriter) error { } // CollectAndSendMonitor 采集并发送监控数据 -func (m *Manager) CollectAndSendMonitor(conn WebSocketWriter) error { - monitorDataList, err := m.monitorCollector.Collect() - if err != nil || len(monitorDataList) == 0 { - // 监控不是必须的,失败或无数据时直接返回 - return nil - } - +func (m *Manager) CollectAndSendMonitor(conn WebSocketWriter, items []protocol.MonitorItem) error { + monitorDataList := m.monitorCollector.Collect(items) return m.sendMetrics(conn, protocol.MetricTypeMonitor, monitorDataList) } diff --git a/pkg/agent/collector/monitor.go b/pkg/agent/collector/monitor.go index f364a37..61a6b82 100644 --- a/pkg/agent/collector/monitor.go +++ b/pkg/agent/collector/monitor.go @@ -11,17 +11,15 @@ import ( "time" "github.com/dushixiang/pika/internal/protocol" - "github.com/dushixiang/pika/pkg/agent/config" ) // MonitorCollector 监控采集器 type MonitorCollector struct { - cfg *config.Config httpClient *http.Client } // NewMonitorCollector 创建监控采集器 -func NewMonitorCollector(cfg *config.Config) *MonitorCollector { +func NewMonitorCollector() *MonitorCollector { // 创建自定义的 HTTP 客户端，支持跳过 TLS 验证 httpClient := &http.Client{ Transport: &http.Transport{ @@ -40,24 +38,19 @@ func NewMonitorCollector(cfg *config.Config) *MonitorCollector { } return &MonitorCollector{ - cfg: cfg, httpClient: httpClient, } } // Collect 采集所有监控项数据 -func (c *MonitorCollector) Collect() ([]protocol.MonitorData, error) { - if !c.cfg.Monitor.Enabled { - return nil, nil +func (c *MonitorCollector) Collect(items []protocol.MonitorItem) []protocol.MonitorData { + if len(items) == 0 { + return nil } - if len(c.cfg.Monitor.Items) == 0 { - return nil, nil - } + results := make([]protocol.MonitorData, 0, len(items)) - results := make([]protocol.MonitorData, 0, len(c.cfg.Monitor.Items)) - - for _, item := range c.cfg.Monitor.Items { + for _, item := range items { var result protocol.MonitorData switch strings.ToLower(item.Type) { @@ -79,11 +72,11 @@ func (c *MonitorCollector) Collect() ([]protocol.MonitorData, error) { results = append(results, result) } - return results, nil + return results } // checkHTTP 检查 HTTP/HTTPS 服务 -func (c *MonitorCollector) checkHTTP(item config.MonitorItem) protocol.MonitorData { +func (c *MonitorCollector) checkHTTP(item protocol.MonitorItem) protocol.MonitorData { result := protocol.MonitorData{ Name: item.Name, Type: item.Type, @@ -94,7 +87,7 @@ func (c *MonitorCollector) checkHTTP(item config.MonitorItem) protocol.MonitorDa // 获取配置，使用默认值 httpCfg := item.HTTPConfig if httpCfg == nil { - httpCfg = &config.HTTPMonitorConfig{ + httpCfg = &protocol.HTTPMonitorConfig{ Method: "GET", ExpectedStatusCode: 200, Timeout: 60, @@ -184,6 +177,20 @@ func (c *MonitorCollector) checkHTTP(item config.MonitorItem) protocol.MonitorDa result.ContentMatch = true } + // 获取 HTTPS 证书信息 + if resp.TLS != nil && len(resp.TLS.PeerCertificates) > 0 { + // 获取第一个证书（服务器证书） + cert := resp.TLS.PeerCertificates[0] + + // 证书过期时间 + expiryTime := cert.NotAfter + result.CertExpiryTime = expiryTime.UnixMilli() + + // 计算剩余天数 + daysLeft := int(time.Until(expiryTime).Hours() / 24) + result.CertDaysLeft = daysLeft + } + // 检查成功 result.Status = "up" result.Message = fmt.Sprintf("HTTP %d - %dms", resp.StatusCode, responseTime) @@ -191,7 +198,7 @@ func (c *MonitorCollector) checkHTTP(item config.MonitorItem) protocol.MonitorDa } // checkTCP 检查 TCP 端口 -func (c *MonitorCollector) checkTCP(item config.MonitorItem) protocol.MonitorData { +func (c *MonitorCollector) checkTCP(item protocol.MonitorItem) protocol.MonitorData { result := protocol.MonitorData{ Name: item.Name, Type: item.Type, diff --git a/pkg/agent/config/config.go b/pkg/agent/config/config.go index a2b4f60..526c6e2 100644 --- a/pkg/agent/config/config.go +++ b/pkg/agent/config/config.go @@ -28,9 +28,6 @@ type Config struct { // 自动更新配置 AutoUpdate AutoUpdateConfig `yaml:"auto_update"` - - // 监控配置 - Monitor MonitorConfig `yaml:"monitor"` } // ServerConfig 服务器配置 @@ -73,65 +70,6 @@ type AutoUpdateConfig struct { CheckInterval string `yaml:"check_interval"` } -// MonitorConfig 监控配置 -type MonitorConfig struct { - // 是否启用监控功能 - Enabled bool `yaml:"enabled"` - - // 检测间隔（秒），默认 60 秒 - Interval int `yaml:"interval"` - - // 监控项列表 - Items []MonitorItem `yaml:"items"` -} - -// MonitorItem 监控项 -type MonitorItem struct { - // 监控项名称 - Name string `yaml:"name"` - - // 监控类型: http, tcp - Type string `yaml:"type"` - - // 目标地址 - // HTTP: 完整的 URL (如: https://example.com/api/health) - // TCP: 地址:端口 (如: example.com:3306) - Target string `yaml:"target"` - - // HTTP 特定配置 - HTTPConfig *HTTPMonitorConfig `yaml:"http_config,omitempty"` - - // TCP 特定配置 - TCPConfig *TCPMonitorConfig `yaml:"tcp_config,omitempty"` -} - -// HTTPMonitorConfig HTTP 监控配置 -type HTTPMonitorConfig struct { - // HTTP 方法: GET, POST, PUT, DELETE 等，默认 GET - Method string `yaml:"method"` - - // 期望的状态码，默认 200 - ExpectedStatusCode int `yaml:"expected_status_code"` - - // 期望的响应内容（关键字） - ExpectedContent string `yaml:"expected_content,omitempty"` - - // 请求超时（秒），默认 60 - Timeout int `yaml:"timeout"` - - // 请求头 - Headers map[string]string `yaml:"headers,omitempty"` - - // 请求体 - Body string `yaml:"body,omitempty"` -} - -// TCPMonitorConfig TCP 监控配置 -type TCPMonitorConfig struct { - // 连接超时（秒），默认 5 - Timeout int `yaml:"timeout"` -} - // DefaultConfig 返回默认配置 func DefaultConfig() *Config { return &Config{ @@ -151,11 +89,6 @@ func DefaultConfig() *Config { Enabled: true, CheckInterval: "10m", }, - Monitor: MonitorConfig{ - Enabled: false, - Interval: 60, - Items: []MonitorItem{}, - }, } } @@ -357,11 +290,3 @@ func (c *Config) ShouldExcludeNetworkInterface(interfaceName string) bool { return false } - -// GetMonitorInterval 获取监控检测间隔时长 -func (c *Config) GetMonitorInterval() time.Duration { - if c.Monitor.Interval <= 0 { - return 60 * time.Second // 默认 60 秒 - } - return time.Duration(c.Monitor.Interval) * time.Second -} diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index bf18997..229aebb 100644 --- a/pkg/agent/service/agent.go +++ b/pkg/agent/service/agent.go @@ -53,9 +53,13 @@ func (sc *safeConn) Close() error { // Agent 探针服务 type Agent struct { - cfg *config.Config - idMgr *id.Manager - cancel context.CancelFunc + cfg *config.Config + idMgr *id.Manager + cancel context.CancelFunc + connMu sync.RWMutex + activeConn *safeConn + collectorMu sync.RWMutex + collectorManager *collector.Manager } // New 创建 Agent 实例 @@ -157,9 +161,16 @@ func (a *Agent) runOnce(ctx context.Context) error { // 创建采集器管理器 collectorManager := collector.NewManager(a.cfg) + a.setActiveConn(conn) + a.setCollectorManager(collectorManager) + defer func() { + a.setCollectorManager(nil) + a.setActiveConn(nil) + }() + // 创建完成通道 done := make(chan struct{}) - errChan := make(chan error, 4) + errChan := make(chan error, 3) // 启动读取循环（处理服务端的 Ping/Pong 等控制消息） go func() { @@ -181,13 +192,6 @@ func (a *Agent) runOnce(ctx context.Context) error { } }() - // 启动监控检测循环 - go func() { - if err := a.monitorLoop(ctx, conn, collectorManager, done); err != nil { - errChan <- fmt.Errorf("监控检测失败: %w", err) - } - }() - // 等待错误或上下文取消 select { case err := <-errChan: @@ -233,9 +237,13 @@ func (a *Agent) readLoop(conn *websocket.Conn, done chan struct{}) error { continue } - // 处理指令消息 - if msg.Type == protocol.MessageTypeCommand { - go a.handleCommand(conn, msg.Data) + switch msg.Type { + case protocol.MessageTypeCommand: + go a.handleCommand(msg.Data) + case protocol.MessageTypeMonitorConfig: + go a.handleMonitorConfig(msg.Data) + default: + // 忽略其他类型 } } } @@ -316,6 +324,35 @@ func (a *Agent) registerAgent(conn *safeConn) error { return nil } +func (a *Agent) handleMonitorConfig(data json.RawMessage) { + var payload protocol.MonitorConfigPayload + if err := json.Unmarshal(data, &payload); err != nil { + log.Printf("⚠️ 解析监控配置失败: %v", err) + return + } + + if len(payload.Items) == 0 { + log.Println("ℹ️ 收到空的服务监控配置，跳过") + return + } + + conn := a.getActiveConn() + manager := a.getCollectorManager() + if conn == nil || manager == nil { + log.Println("⚠️ 当前连接未就绪，无法执行服务监控任务") + return + } + + log.Printf("📥 收到服务监控配置，总计 %d 个监控项，立即执行检测", len(payload.Items)) + + // 立即执行一次监控检测 + if err := manager.CollectAndSendMonitor(conn, payload.Items); err != nil { + log.Printf("⚠️ 监控检测失败: %v", err) + } else { + log.Printf("✅ 服务监控检测完成，已上报 %d 个监控项结果", len(payload.Items)) + } +} + // heartbeatLoop 心跳循环 func (a *Agent) heartbeatLoop(ctx context.Context, conn *safeConn, done chan struct{}) error { ticker := time.NewTicker(a.cfg.GetHeartbeatInterval()) @@ -340,6 +377,30 @@ func (a *Agent) heartbeatLoop(ctx context.Context, conn *safeConn, done chan str } } +func (a *Agent) setActiveConn(conn *safeConn) { + a.connMu.Lock() + defer a.connMu.Unlock() + a.activeConn = conn +} + +func (a *Agent) getActiveConn() *safeConn { + a.connMu.RLock() + defer a.connMu.RUnlock() + return a.activeConn +} + +func (a *Agent) setCollectorManager(manager *collector.Manager) { + a.collectorMu.Lock() + defer a.collectorMu.Unlock() + a.collectorManager = manager +} + +func (a *Agent) getCollectorManager() *collector.Manager { + a.collectorMu.RLock() + defer a.collectorMu.RUnlock() + return a.collectorManager +} + // metricsLoop 指标采集循环 func (a *Agent) metricsLoop(ctx context.Context, conn *safeConn, manager *collector.Manager, done chan struct{}) error { // 立即采集一次动态数据 @@ -434,47 +495,8 @@ func (a *Agent) collectAndSendAllMetrics(conn *safeConn, manager *collector.Mana return nil } -// monitorLoop 监控检测循环 -func (a *Agent) monitorLoop(ctx context.Context, conn *safeConn, manager *collector.Manager, done chan struct{}) error { - // 如果监控未启用，直接返回 - if !a.cfg.Monitor.Enabled { - log.Println("ℹ️ 监控功能未启用") - return nil - } - - if len(a.cfg.Monitor.Items) == 0 { - log.Println("ℹ️ 没有配置监控项") - return nil - } - - log.Printf("🔍 监控功能已启用，共 %d 个监控项", len(a.cfg.Monitor.Items)) - - // 立即执行一次监控检测 - if err := manager.CollectAndSendMonitor(conn); err != nil { - log.Printf("⚠️ 初始监控检测失败: %v", err) - } - - // 定时监控检测 - ticker := time.NewTicker(a.cfg.GetMonitorInterval()) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - // 采集并发送监控数据 - if err := manager.CollectAndSendMonitor(conn); err != nil { - log.Printf("⚠️ 监控检测失败: %v", err) - } - case <-done: - return nil - case <-ctx.Done(): - return nil - } - } -} - // handleCommand 处理服务端下发的指令 -func (a *Agent) handleCommand(conn *websocket.Conn, data json.RawMessage) { +func (a *Agent) handleCommand(data json.RawMessage) { var cmdReq protocol.CommandRequest if err := json.Unmarshal(data, &cmdReq); err != nil { log.Printf("⚠️ 解析指令失败: %v", err) @@ -483,6 +505,7 @@ func (a *Agent) handleCommand(conn *websocket.Conn, data json.RawMessage) { log.Printf("📥 收到指令: %s (ID: %s)", cmdReq.Type, cmdReq.ID) + conn := a.getActiveConn() // 发送运行中状态 a.sendCommandResponse(conn, cmdReq.ID, cmdReq.Type, "running", "", "") @@ -496,7 +519,7 @@ func (a *Agent) handleCommand(conn *websocket.Conn, data json.RawMessage) { } // handleVPSAudit 处理VPS安全审计指令 -func (a *Agent) handleVPSAudit(conn *websocket.Conn, cmdID string) { +func (a *Agent) handleVPSAudit(conn *safeConn, cmdID string) { // 导入 audit 包 result, err := a.runVPSAudit() if err != nil { @@ -523,7 +546,7 @@ func (a *Agent) runVPSAudit() (*protocol.VPSAuditResult, error) { } // sendCommandResponse 发送指令响应 -func (a *Agent) sendCommandResponse(conn *websocket.Conn, cmdID, cmdType, status, errMsg, result string) { +func (a *Agent) sendCommandResponse(conn *safeConn, cmdID, cmdType, status, errMsg, result string) { resp := protocol.CommandResponse{ ID: cmdID, Type: cmdType, diff --git a/web/src/api/agent.ts b/web/src/api/agent.ts index b7155b1..c7dabd1 100644 --- a/web/src/api/agent.ts +++ b/web/src/api/agent.ts @@ -9,7 +9,7 @@ export interface ListAgentsResponse { export interface GetAgentMetricsRequest { agentId: string; type: 'cpu' | 'memory' | 'disk' | 'network' | 'load' | 'disk_io' | 'gpu' | 'temperature'; - range?: '1h' | '6h' | '12h' | '24h' | '1d' | '3d' | '7d' | '1w' | '30d' | '1M' | '90d' | '3M'; + range?: '1m' | '5m' | '15m' | '30m' | '1h'; } export interface GetAgentMetricsResponse { @@ -184,4 +184,3 @@ export interface VersionInfo { export const getServerVersion = () => { return get('/agent/version'); }; - diff --git a/web/src/api/monitor.ts b/web/src/api/monitor.ts new file mode 100644 index 0000000..4121089 --- /dev/null +++ b/web/src/api/monitor.ts @@ -0,0 +1,57 @@ +import {get, post, put, del} from './request'; +import type {MonitorListResponse, MonitorTask, MonitorTaskRequest, MonitorStats} from '../types'; + +export const listMonitors = (page: number = 1, pageSize: number = 10, keyword?: string) => { + const params = new URLSearchParams(); + params.append('page', page.toString()); + params.append('pageSize', pageSize.toString()); + if (keyword) { + params.append('keyword', keyword); + } + return get(`/admin/monitors?${params.toString()}`); +}; + +export const createMonitor = (data: MonitorTaskRequest) => { + return post('/admin/monitors', data); +}; + +export const getMonitor = (id: number) => { + return get(`/admin/monitors/${id}`); +}; + +export const updateMonitor = (id: number, data: MonitorTaskRequest) => { + return put(`/admin/monitors/${id}`, data); +}; + +export const deleteMonitor = (id: number) => { + return del(`/admin/monitors/${id}`); +}; + +// 公开接口 - 获取所有监控统计数据 +export const getAllMonitorStats = () => { + return get('/monitors/stats'); +}; + +// 公开接口 - 获取指定监控的统计数据 +export const getMonitorStatsByName = (name: string) => { + return get(`/monitors/${encodeURIComponent(name)}/stats`); +}; + +// 聚合的监控历史数据 +export interface AggregatedMonitorMetric { + timestamp: number; + agentId: string; + avgResponse: number; + maxResponse: number; + minResponse: number; + successCount: number; + totalCount: number; + successRate: number; + lastStatus: string; + lastErrorMsg: string; +} + +// 公开接口 - 获取指定监控的历史数据 +export const getMonitorHistory = (name: string, range: string = '5m') => { + return get(`/monitors/${encodeURIComponent(name)}/history?range=${range}`); +}; diff --git a/web/src/components/PublicFooter.tsx b/web/src/components/PublicFooter.tsx new file mode 100644 index 0000000..4d8169f --- /dev/null +++ b/web/src/components/PublicFooter.tsx @@ -0,0 +1,11 @@ +const PublicFooter = () => { + return ( +
+
+ © {new Date().getFullYear()} Pika Monitor · 保持洞察，稳定运行。 +
+
+ ); +}; + +export default PublicFooter; diff --git a/web/src/components/PublicHeader.tsx b/web/src/components/PublicHeader.tsx new file mode 100644 index 0000000..3a8eda1 --- /dev/null +++ b/web/src/components/PublicHeader.tsx @@ -0,0 +1,119 @@ +import {Activity, LayoutGrid, List, LogIn, Server} from 'lucide-react'; +import GithubSvg from "../assets/github.svg"; + +interface PublicHeaderProps { + title: string; + lastUpdated?: string; + viewMode?: 'grid' | 'list'; + onViewModeChange?: (mode: 'grid' | 'list') => void; + showViewToggle?: boolean; +} + +const PublicHeader = ({ + title, + lastUpdated, + viewMode, + onViewModeChange, + showViewToggle = false + }: PublicHeaderProps) => { + return ( +
+
+
+ {/* 左侧：标题 */} +
+

Pika Monitor

+

{title}

+
+ + {/* 右侧：操作区域 */} +
+ {/* 最后更新时间 */} + {lastUpdated && ( + <> + + 最后更新： + {lastUpdated} + + + + )} + + {/* 视图切换 */} + {showViewToggle && viewMode && onViewModeChange && ( + <> +
+ + +
+ + + )} + + {/* GitHub 链接 */} + + github + + + {/* 服务器链接 */} + + + 服务器 + + + {/* 监控链接 */} + + + 服务监控 + + + {/* 登录按钮 */} + + + 登录 + +
+
+
+
+ ); +}; + +export default PublicHeader; diff --git a/web/src/pages/AdminLayout/index.tsx b/web/src/pages/AdminLayout/index.tsx index e85a547..d9e395e 100644 --- a/web/src/pages/AdminLayout/index.tsx +++ b/web/src/pages/AdminLayout/index.tsx @@ -2,7 +2,7 @@ import { type JSX, useEffect, useMemo, useState } from 'react'; import { Outlet, useLocation, useNavigate } from 'react-router-dom'; import type { MenuProps } from 'antd'; import { App, Avatar, Button, Dropdown, Space } from 'antd'; -import { Eye, Key, LogOut, Server, User as UserIcon, Users, BookOpen, Settings } from 'lucide-react'; +import { Eye, Key, LogOut, Server, User as UserIcon, Users, BookOpen, Settings, Activity } from 'lucide-react'; import { logout } from '../../api/auth'; import { getServerVersion } from '../../api/agent'; import type { User } from '../../types'; @@ -40,6 +40,12 @@ const AdminLayout = () => { path: '/admin/api-keys', icon: , }, + { + key: 'monitors', + label: '服务监控', + path: '/admin/monitors', + icon: , + }, { key: 'users', label: '用户管理', @@ -72,6 +78,8 @@ const AdminLayout = () => { setSelectedKey('users'); } else if (path.startsWith('/admin/api-keys')) { setSelectedKey('api-keys'); + } else if (path.startsWith('/admin/monitors')) { + setSelectedKey('monitors'); } else if (path.startsWith('/admin/settings')) { setSelectedKey('settings'); } else { @@ -151,7 +159,7 @@ const AdminLayout = () => { type="text" icon={} onClick={() => navigate('/admin/agents-install')} - className="!h-9 !items-center !rounded-full !px-3 !text-xs !text-white hover:!bg-indigo-500/10" + className="!h-9 !items-center !rounded-full !px-3 !text-xs !text-white hover:!bg-blue-500/10" > 部署指南 @@ -200,20 +208,20 @@ const AdminLayout = () => { className={cn( 'group relative flex w-full items-center gap-3 rounded-2xl px-3 py-2 text-sm transition-all cursor-pointer', isActive - ? 'bg-gradient-to-r from-indigo-500/10 to-indigo-500/5 text-indigo-600 shadow-sm' + ? 'bg-gradient-to-r from-blue-500/10 to-blue-500/5 text-blue-600 shadow-sm' : 'text-gray-600 hover:bg-gray-100' )} > {item.icon} {item.label} - {isActive && 当前} + {isActive && 当前} ); })} @@ -249,7 +257,7 @@ const AdminLayout = () => { {/* 移动端底部导航栏 */}