okooooooo/pika
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加自定义采集数据保留时长

Browse Source
This commit is contained in:
dushixiang committed 2025-11-29 23:14:42 +08:00
1 parent 173565b15c
commit b82e982c84
18 files changed
+1222 -250
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/app.go
+13 -1
View File
@@ -83,6 +83,9 @@ func setup(app *orz.App) error {
// 启动数据清理任务
go components.AgentService.StartCleanupTask(ctx)
// 启动聚合下采样任务
go components.AgentService.StartAggregationTask(ctx)
// 启动指标监控任务（用于告警检测）
go startMetricsMonitoring(ctx, components, app.Logger())
@@ -274,7 +277,6 @@ func autoMigrate(database *gorm.DB) error {
&models.DiskMetric{},
&models.NetworkMetric{},
&models.NetworkConnectionMetric{},
&models.LoadMetric{},
&models.DiskIOMetric{},
&models.GPUMetric{},
&models.TemperatureMetric{},
@@ -289,6 +291,16 @@ func autoMigrate(database *gorm.DB) error {
&models.TamperProtectConfig{},
&models.TamperEvent{},
&models.TamperAlert{},
// 聚合表
&models.AggregatedCPUMetricModel{},
&models.AggregatedMemoryMetricModel{},
&models.AggregatedDiskMetricModel{},
&models.AggregatedNetworkMetricModel{},
&models.AggregatedNetworkConnectionMetricModel{},
&models.AggregatedDiskIOMetricModel{},
&models.AggregatedGPUMetricModel{},
&models.AggregatedTemperatureMetricModel{},
&models.AggregationProgress{},
)
}
internal/handler/agent_handler.go
+59 -70
View File
@@ -354,33 +354,8 @@ func (h *AgentHandler) GetForAdmin(c echo.Context) error {
return orz.Ok(c, agent)
}
// GetMetrics 获取探针聚合指标（公开接口，已登录返回全部，未登录返回公开可见）
func (h *AgentHandler) GetMetrics(c echo.Context) error {
agentID := c.Param("id")
ctx := c.Request().Context()
// 验证探针访问权限
if _, err := h.agentService.GetAgentByAuth(ctx, agentID, utils.IsAuthenticated(c)); err != nil {
return err
}
metricType := c.QueryParam("type")
rangeParam := c.QueryParam("range")
// 验证指标类型
validTypes := map[string]bool{
"cpu": true, "memory": true, "disk": true, "network": true, "network_connection": true, "load": true,
"disk_io": true, "gpu": true, "temperature": true,
}
if metricType == "" {
return orz.NewError(400, "指标类型不能为空")
}
if !validTypes[metricType] {
return orz.NewError(400, "无效的指标类型")
}
// 根据 range 参数自动计算时间范围
var start, end int64
// parseTimeRange 解析时间范围参数，返回起始和结束时间（毫秒）
func parseTimeRange(rangeParam string) (start, end int64, err error) {
end = time.Now().UnixMilli()
if rangeParam == "" {
@@ -398,8 +373,56 @@ func (h *AgentHandler) GetMetrics(c echo.Context) error {
start = end - 30*60*1000
case "1h":
start = end - 60*60*1000
case "3h":
start = end - 3*60*60*1000
case "6h":
start = end - 6*60*60*1000
case "12h":
start = end - 12*60*60*1000
case "1d", "24h":
start = end - 24*60*60*1000
case "3d":
start = end - 3*24*60*60*1000
case "7d":
start = end - 7*24*60*60*1000
case "30d":
start = end - 30*24*60*60*1000
default:
return orz.NewError(400, "无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h")
return 0, 0, fmt.Errorf("无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h, 3h, 6h, 12h, 1d/24h, 3d, 7d, 30d")
}
return start, end, nil
}
// GetMetrics 获取探针聚合指标（公开接口，已登录返回全部，未登录返回公开可见）
func (h *AgentHandler) GetMetrics(c echo.Context) error {
agentID := c.Param("id")
ctx := c.Request().Context()
// 验证探针访问权限
if _, err := h.agentService.GetAgentByAuth(ctx, agentID, utils.IsAuthenticated(c)); err != nil {
return err
}
metricType := c.QueryParam("type")
rangeParam := c.QueryParam("range")
// 验证指标类型
validTypes := map[string]bool{
"cpu": true, "memory": true, "disk": true, "network": true, "network_connection": true,
"disk_io": true, "gpu": true, "temperature": true,
}
if metricType == "" {
return orz.NewError(400, "指标类型不能为空")
}
if !validTypes[metricType] {
return orz.NewError(400, "无效的指标类型")
}
// 解析时间范围
start, end, err := parseTimeRange(rangeParam)
if err != nil {
return orz.NewError(400, err.Error())
}
// 服务端自动计算最优聚合间隔
@@ -433,27 +456,10 @@ func (h *AgentHandler) GetNetworkMetricsByInterface(c echo.Context) error {
rangeParam := c.QueryParam("range")
// 根据 range 参数自动计算时间范围
var start, end int64
end = time.Now().UnixMilli()
if rangeParam == "" {
rangeParam = "1h" // 默认1小时
}
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 - 60*60*1000
default:
return orz.NewError(400, "无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h")
// 解析时间范围
start, end, err := parseTimeRange(rangeParam)
if err != nil {
return orz.NewError(400, err.Error())
}
// 服务端自动计算最优聚合间隔
@@ -701,27 +707,10 @@ func (h *AgentHandler) GetMonitorMetrics(c echo.Context) error {
rangeParam := c.QueryParam("range")
ctx := c.Request().Context()
// 根据 range 参数自动计算时间范围
var start, end int64
end = time.Now().UnixMilli()
if rangeParam == "" {
rangeParam = "1h" // 默认1小时
}
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 - 60*60*1000
default:
return orz.NewError(400, "无效的时间范围，支持: 1m, 5m, 15m, 30m, 1h")
// 解析时间范围
start, end, err := parseTimeRange(rangeParam)
if err != nil {
return orz.NewError(400, err.Error())
}
metrics, err := h.agentService.GetMonitorMetrics(ctx, agentID, monitorName, start, end)
internal/models/metric.go
+147 -14
View File
@@ -88,20 +88,6 @@ func (NetworkConnectionMetric) TableName() string {
return "network_connection_metrics"
}
// LoadMetric 系统负载指标
type LoadMetric struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_load_agent_ts,priority:1" json:"agentId"` // 探针ID
Load1 float64 `json:"load1"` // 1分钟负载
Load5 float64 `json:"load5"` // 5分钟负载
Load15 float64 `json:"load15"` // 15分钟负载
Timestamp int64 `gorm:"index:idx_load_agent_ts,priority:2;index:idx_load_ts" json:"timestamp"` // 时间戳（毫秒）
}
func (LoadMetric) TableName() string {
return "load_metrics"
}
// DiskIOMetric 磁盘IO指标
type DiskIOMetric struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
@@ -199,3 +185,150 @@ type MonitorMetric struct {
func (MonitorMetric) TableName() string {
return "monitor_metrics"
}
// ----------- 聚合表 -----------
// AggregatedCPUMetricModel CPU聚合表
type AggregatedCPUMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_cpuagg_agent_bucket,priority:1;uniqueIndex:ux_cpuagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_cpuagg_agent_bucket,priority:2;uniqueIndex:ux_cpuagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_cpuagg_agent_bucket,priority:3;uniqueIndex:ux_cpuagg_bucket,priority:3" json:"bucketStart"` // 毫秒
MaxUsage float64 `json:"maxUsage"`
LogicalCores int `json:"logicalCores"`
}
func (AggregatedCPUMetricModel) TableName() string {
return "cpu_metrics_aggs"
}
// AggregatedMemoryMetricModel 内存聚合表
type AggregatedMemoryMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_memagg_agent_bucket,priority:1;uniqueIndex:ux_memagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_memagg_agent_bucket,priority:2;uniqueIndex:ux_memagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_memagg_agent_bucket,priority:3;uniqueIndex:ux_memagg_bucket,priority:3" json:"bucketStart"` // 毫秒
MaxUsage float64 `json:"maxUsage"`
Total uint64 `json:"total"`
}
func (AggregatedMemoryMetricModel) TableName() string {
return "memory_metrics_aggs"
}
// AggregatedDiskMetricModel 磁盘聚合表
type AggregatedDiskMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_diskagg_agent_bucket_mp,priority:1;uniqueIndex:ux_diskagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_diskagg_agent_bucket_mp,priority:2;uniqueIndex:ux_diskagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_diskagg_agent_bucket_mp,priority:3;uniqueIndex:ux_diskagg_bucket,priority:3" json:"bucketStart"` // 毫秒
MountPoint string `gorm:"index:idx_diskagg_agent_bucket_mp,priority:4;uniqueIndex:ux_diskagg_bucket,priority:4" json:"mountPoint"`
MaxUsage float64 `json:"maxUsage"`
Total uint64 `json:"total"`
}
func (AggregatedDiskMetricModel) TableName() string {
return "disk_metrics_aggs"
}
// AggregatedNetworkMetricModel 网络聚合表（按网卡分组）
type AggregatedNetworkMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_netagg_agent_bucket_iface,priority:1;uniqueIndex:ux_netagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_netagg_agent_bucket_iface,priority:2;uniqueIndex:ux_netagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_netagg_agent_bucket_iface,priority:3;uniqueIndex:ux_netagg_bucket,priority:3" json:"bucketStart"` // 毫秒
Interface string `gorm:"index:idx_netagg_agent_bucket_iface,priority:4;uniqueIndex:ux_netagg_bucket,priority:4" json:"interface"`
MaxSentRate float64 `json:"maxSentRate"`
MaxRecvRate float64 `json:"maxRecvRate"`
}
func (AggregatedNetworkMetricModel) TableName() string {
return "network_metrics_aggs"
}
// AggregatedNetworkConnectionMetricModel 网络连接聚合表
type AggregatedNetworkConnectionMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_netconnagg_agent_bucket,priority:1;uniqueIndex:ux_netconnagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_netconnagg_agent_bucket,priority:2;uniqueIndex:ux_netconnagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_netconnagg_agent_bucket,priority:3;uniqueIndex:ux_netconnagg_bucket,priority:3" json:"bucketStart"` // 毫秒
MaxEstablished uint32 `json:"maxEstablished"`
MaxSynSent uint32 `json:"maxSynSent"`
MaxSynRecv uint32 `json:"maxSynRecv"`
MaxFinWait1 uint32 `json:"maxFinWait1"`
MaxFinWait2 uint32 `json:"maxFinWait2"`
MaxTimeWait uint32 `json:"maxTimeWait"`
MaxClose uint32 `json:"maxClose"`
MaxCloseWait uint32 `json:"maxCloseWait"`
MaxLastAck uint32 `json:"maxLastAck"`
MaxListen uint32 `json:"maxListen"`
MaxClosing uint32 `json:"maxClosing"`
MaxTotal uint32 `json:"maxTotal"`
}
func (AggregatedNetworkConnectionMetricModel) TableName() string {
return "network_connection_metrics_aggs"
}
// AggregatedDiskIOMetricModel 磁盘IO聚合表
type AggregatedDiskIOMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_diskioagg_agent_bucket_dev,priority:1;uniqueIndex:ux_diskioagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_diskioagg_agent_bucket_dev,priority:2;uniqueIndex:ux_diskioagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_diskioagg_agent_bucket_dev,priority:3;uniqueIndex:ux_diskioagg_bucket,priority:3" json:"bucketStart"` // 毫秒
Device string `gorm:"index:idx_diskioagg_agent_bucket_dev,priority:4;uniqueIndex:ux_diskioagg_bucket,priority:4" json:"device"`
MaxReadBytesRate uint64 `json:"maxReadBytesRate"`
MaxWriteBytesRate uint64 `json:"maxWriteBytesRate"`
MaxIopsInProgress uint64 `json:"maxIopsInProgress"`
}
func (AggregatedDiskIOMetricModel) TableName() string {
return "disk_io_metrics_aggs"
}
// AggregatedGPUMetricModel GPU聚合表
type AggregatedGPUMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_gpuagg_agent_bucket_idx,priority:1;uniqueIndex:ux_gpuagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_gpuagg_agent_bucket_idx,priority:2;uniqueIndex:ux_gpuagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_gpuagg_agent_bucket_idx,priority:3;uniqueIndex:ux_gpuagg_bucket,priority:3" json:"bucketStart"` // 毫秒
Index int `gorm:"index:idx_gpuagg_agent_bucket_idx,priority:4;uniqueIndex:ux_gpuagg_bucket,priority:4" json:"index"`
Name string `json:"name"`
MaxUtilization float64 `json:"maxUtilization"`
MaxMemoryUsed uint64 `json:"maxMemoryUsed"`
MaxTemperature float64 `json:"maxTemperature"`
MaxPowerDraw float64 `json:"maxPowerDraw"`
MemoryTotal uint64 `json:"memoryTotal"`
}
func (AggregatedGPUMetricModel) TableName() string {
return "gpu_metrics_aggs"
}
// AggregatedTemperatureMetricModel 温度聚合表
type AggregatedTemperatureMetricModel struct {
ID uint `gorm:"primaryKey;autoIncrement" json:"id"`
AgentID string `gorm:"index:idx_tempagg_agent_bucket_sensor,priority:1;uniqueIndex:ux_tempagg_bucket,priority:1" json:"agentId"`
BucketSeconds int `gorm:"index:idx_tempagg_agent_bucket_sensor,priority:2;uniqueIndex:ux_tempagg_bucket,priority:2" json:"bucketSeconds"`
BucketStart int64 `gorm:"index:idx_tempagg_agent_bucket_sensor,priority:3;uniqueIndex:ux_tempagg_bucket,priority:3" json:"bucketStart"` // 毫秒
SensorKey string `gorm:"index:idx_tempagg_agent_bucket_sensor,priority:4;uniqueIndex:ux_tempagg_bucket,priority:4" json:"sensorKey"`
SensorLabel string `json:"sensorLabel"`
MaxTemperature float64 `json:"maxTemperature"`
}
func (AggregatedTemperatureMetricModel) TableName() string {
return "temperature_metrics_aggs"
}
// AggregationProgress 聚合进度记录
type AggregationProgress struct {
MetricType string `gorm:"primaryKey"` // cpu/memory/disk/network
BucketSeconds int `gorm:"primaryKey"`
LastBucket int64 `json:"lastBucket"` // 已完成的最后 bucket 起始时间（毫秒）
UpdatedAt int64 `json:"updatedAt" gorm:"autoUpdateTime:milli"`
CreatedAt int64 `json:"createdAt" gorm:"autoCreateTime:milli"`
}
func (AggregationProgress) TableName() string {
return "aggregation_progress"
}
internal/models/property.go
+6
View File
@@ -48,3 +48,9 @@ type SystemConfig struct {
ICPCode string `json:"icpCode"` // ICP备案号
DefaultView string `json:"defaultView"` // 默认视图 grid | list
}
// MetricsConfig 指标数据配置
type MetricsConfig struct {
RetentionHours int `json:"retentionHours"` // 原始数据保留小时数（默认168小时=7天）
MaxQueryPoints int `json:"maxQueryPoints"` // 查询返回的最大数据点数（默认720）
}
internal/protocol/message.go
-1
View File
@@ -66,7 +66,6 @@ const (
MetricTypeDiskIO MetricType = "disk_io"
MetricTypeNetwork MetricType = "network"
MetricTypeNetworkConnection MetricType = "network_connection"
MetricTypeLoad MetricType = "load"
MetricTypeHost MetricType = "host"
MetricTypeGPU MetricType = "gpu"
MetricTypeTemperature MetricType = "temperature"
internal/repo/metric_repo.go
+357 -59
View File
@@ -39,11 +39,6 @@ func (r *MetricRepo) SaveNetworkMetric(ctx context.Context, metric *models.Netwo
return r.db.WithContext(ctx).Create(metric).Error
}
// SaveLoadMetric 保存负载指标
func (r *MetricRepo) SaveLoadMetric(ctx context.Context, metric *models.LoadMetric) error {
return r.db.WithContext(ctx).Create(metric).Error
}
// SaveDiskIOMetric 保存磁盘IO指标
func (r *MetricRepo) SaveDiskIOMetric(ctx context.Context, metric *models.DiskIOMetric) error {
return r.db.WithContext(ctx).Create(metric).Error
@@ -202,7 +197,6 @@ func (r *MetricRepo) DeleteOldMetrics(ctx context.Context, beforeTimestamp int64
&models.DiskMetric{},
&models.NetworkMetric{},
&models.NetworkConnectionMetric{},
&models.LoadMetric{},
&models.DiskIOMetric{},
&models.GPUMetric{},
&models.TemperatureMetric{},
@@ -325,9 +319,10 @@ func (r *MetricRepo) GetDiskMetrics(ctx context.Context, agentID string, start,
return metrics, err
}
// AggregatedNetworkMetric 网络聚合指标（使用最大值）
// AggregatedNetworkMetric 网络聚合指标（按网卡分组）
type AggregatedNetworkMetric struct {
Timestamp int64 `json:"timestamp"`
Interface string `json:"interface"`
MaxSentRate float64 `json:"maxSentRate"`
MaxRecvRate float64 `json:"maxRecvRate"`
}
@@ -387,38 +382,6 @@ func (r *MetricRepo) GetNetworkMetricsByInterface(ctx context.Context, agentID s
return metrics, err
}
// AggregatedLoadMetric 负载聚合指标（使用最大值）
type AggregatedLoadMetric struct {
Timestamp int64 `json:"timestamp"`
MaxLoad1 float64 `json:"maxLoad1"`
MaxLoad5 float64 `json:"maxLoad5"`
MaxLoad15 float64 `json:"maxLoad15"`
}
// GetLoadMetrics 获取聚合后的负载指标（始终返回聚合数据）
func (r *MetricRepo) GetLoadMetrics(ctx context.Context, agentID string, start, end int64, interval int) ([]AggregatedLoadMetric, error) {
var metrics []AggregatedLoadMetric
query := `
SELECT
CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp,
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
ORDER BY timestamp ASC
`
intervalMs := int64(interval * 1000)
err := r.db.WithContext(ctx).
Raw(query, intervalMs, intervalMs, agentID, start, end).
Scan(&metrics).Error
return metrics, err
}
// GetLatestCPUMetric 获取最新的CPU指标
func (r *MetricRepo) GetLatestCPUMetric(ctx context.Context, agentID string) (*models.CPUMetric, error) {
var metric models.CPUMetric
@@ -499,19 +462,6 @@ func (r *MetricRepo) GetLatestNetworkMetrics(ctx context.Context, agentID string
return metrics, err
}
// GetLatestLoadMetric 获取最新的负载指标
func (r *MetricRepo) GetLatestLoadMetric(ctx context.Context, agentID string) (*models.LoadMetric, error) {
var metric models.LoadMetric
err := r.db.WithContext(ctx).
Where("agent_id = ?", agentID).
Order("timestamp DESC").
First(&metric).Error
if err != nil {
return nil, err
}
return &metric, nil
}
// SaveMonitorMetric 保存监控指标
func (r *MetricRepo) SaveMonitorMetric(ctx context.Context, metric *models.MonitorMetric) error {
return r.db.WithContext(ctx).Create(metric).Error
@@ -622,12 +572,13 @@ func (r *MetricRepo) GetAggregatedMonitorMetrics(ctx context.Context, monitorID
// AggregatedDiskIOMetric 磁盘IO聚合指标
type AggregatedDiskIOMetric struct {
Timestamp int64 `json:"timestamp"`
Device string `json:"device"`
MaxReadRate float64 `json:"maxReadRate"` // 最大读取速率(字节/秒)
MaxWriteRate float64 `json:"maxWriteRate"` // 最大写入速率(字节/秒)
TotalReadBytes uint64 `json:"totalReadBytes"` // 总读取字节数
TotalWriteBytes uint64 `json:"totalWriteBytes"` // 总写入字节数
Timestamp int64 `json:"timestamp"`
Device string `json:"device"`
MaxReadRate float64 `json:"maxReadRate"` // 最大读取速率(字节/秒)
MaxWriteRate float64 `json:"maxWriteRate"` // 最大写入速率(字节/秒)
TotalReadBytes uint64 `json:"totalReadBytes"` // 总读取字节数
TotalWriteBytes uint64 `json:"totalWriteBytes"` // 总写入字节数
MaxIopsInProgress uint64 `json:"maxIopsInProgress"` // 最大进行中的IO操作数
}
// GetDiskIOMetrics 获取聚合后的磁盘IO指标
@@ -666,10 +617,13 @@ func (r *MetricRepo) GetDiskIOMetrics(ctx context.Context, agentID string, start
// AggregatedGPUMetric GPU聚合指标（使用最大值）
type AggregatedGPUMetric struct {
Timestamp int64 `json:"timestamp"`
Index int `json:"index"`
Name string `json:"name"`
MaxUtilization float64 `json:"maxUtilization"`
MaxMemoryUsed uint64 `json:"maxMemoryUsed"`
MaxTemperature float64 `json:"maxTemperature"`
MaxPowerDraw float64 `json:"maxPowerDraw"`
MemoryTotal uint64 `json:"memoryTotal"`
}
// GetGPUMetrics 获取聚合后的GPU指标
@@ -744,7 +698,6 @@ func (r *MetricRepo) DeleteAgentMetrics(ctx context.Context, agentID string) err
&models.DiskMetric{},
&models.DiskIOMetric{},
&models.NetworkMetric{},
&models.LoadMetric{},
&models.HostMetric{},
&models.GPUMetric{},
&models.TemperatureMetric{},
@@ -798,3 +751,348 @@ func (r *MetricRepo) GetAllLatestMonitorMetrics(ctx context.Context) ([]*models.
return metrics, err
}
// ----------- 聚合表操作 -----------
// AggregateCPUToAgg 将原始CPU数据聚合到聚合表
func (r *MetricRepo) AggregateCPUToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO cpu_metrics_aggs (agent_id, bucket_seconds, bucket_start, max_usage, logical_cores)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
MAX(usage_percent) as max_usage,
MAX(logical_cores) as logical_cores
FROM cpu_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start
ON CONFLICT (agent_id, bucket_seconds, bucket_start) DO UPDATE SET
max_usage = EXCLUDED.max_usage,
logical_cores = EXCLUDED.logical_cores
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateMemoryToAgg 将原始内存数据聚合到聚合表
func (r *MetricRepo) AggregateMemoryToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO memory_metrics_aggs (agent_id, bucket_seconds, bucket_start, max_usage, total)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
MAX(usage_percent) as max_usage,
MAX(total) as total
FROM memory_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start
ON CONFLICT (agent_id, bucket_seconds, bucket_start) DO UPDATE SET
max_usage = EXCLUDED.max_usage,
total = EXCLUDED.total
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateDiskToAgg 将原始磁盘数据聚合到聚合表
func (r *MetricRepo) AggregateDiskToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO disk_metrics_aggs (agent_id, bucket_seconds, bucket_start, mount_point, max_usage, total)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
mount_point,
MAX(usage_percent) as max_usage,
MAX(total) as total
FROM disk_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start, mount_point
ON CONFLICT (agent_id, bucket_seconds, bucket_start, mount_point) DO UPDATE SET
max_usage = EXCLUDED.max_usage,
total = EXCLUDED.total
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateNetworkToAgg 将原始网络数据聚合到聚合表（按网卡分组）
func (r *MetricRepo) AggregateNetworkToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO network_metrics_aggs (agent_id, bucket_seconds, bucket_start, interface, max_sent_rate, max_recv_rate)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
interface,
MAX(bytes_sent_rate) as max_sent_rate,
MAX(bytes_recv_rate) as max_recv_rate
FROM network_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start, interface
ON CONFLICT (agent_id, bucket_seconds, bucket_start, interface) DO UPDATE SET
max_sent_rate = EXCLUDED.max_sent_rate,
max_recv_rate = EXCLUDED.max_recv_rate
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateNetworkConnectionToAgg 将原始网络连接数据聚合到聚合表
func (r *MetricRepo) AggregateNetworkConnectionToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO network_connection_metrics_aggs (
agent_id, bucket_seconds, bucket_start,
max_established, max_syn_sent, max_syn_recv, max_fin_wait1, max_fin_wait2,
max_time_wait, max_close, max_close_wait, max_last_ack, max_listen,
max_closing, max_total
)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
MAX(established) as max_established,
MAX(syn_sent) as max_syn_sent,
MAX(syn_recv) as max_syn_recv,
MAX(fin_wait1) as max_fin_wait1,
MAX(fin_wait2) as max_fin_wait2,
MAX(time_wait) as max_time_wait,
MAX(close) as max_close,
MAX(close_wait) as max_close_wait,
MAX(last_ack) as max_last_ack,
MAX(listen) as max_listen,
MAX(closing) as max_closing,
MAX(total) as max_total
FROM network_connection_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start
ON CONFLICT (agent_id, bucket_seconds, bucket_start) DO UPDATE SET
max_established = EXCLUDED.max_established,
max_syn_sent = EXCLUDED.max_syn_sent,
max_syn_recv = EXCLUDED.max_syn_recv,
max_fin_wait1 = EXCLUDED.max_fin_wait1,
max_fin_wait2 = EXCLUDED.max_fin_wait2,
max_time_wait = EXCLUDED.max_time_wait,
max_close = EXCLUDED.max_close,
max_close_wait = EXCLUDED.max_close_wait,
max_last_ack = EXCLUDED.max_last_ack,
max_listen = EXCLUDED.max_listen,
max_closing = EXCLUDED.max_closing,
max_total = EXCLUDED.max_total
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateDiskIOToAgg 将原始磁盘IO数据聚合到聚合表
func (r *MetricRepo) AggregateDiskIOToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO disk_io_metrics_aggs (
agent_id, bucket_seconds, bucket_start, device,
max_read_bytes_rate, max_write_bytes_rate, max_iops_in_progress
)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
device,
MAX(read_bytes_rate) as max_read_bytes_rate,
MAX(write_bytes_rate) as max_write_bytes_rate,
MAX(iops_in_progress) as max_iops_in_progress
FROM disk_io_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start, device
ON CONFLICT (agent_id, bucket_seconds, bucket_start, device) DO UPDATE SET
max_read_bytes_rate = EXCLUDED.max_read_bytes_rate,
max_write_bytes_rate = EXCLUDED.max_write_bytes_rate,
max_iops_in_progress = EXCLUDED.max_iops_in_progress
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateGPUToAgg 将原始GPU数据聚合到聚合表
func (r *MetricRepo) AggregateGPUToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO gpu_metrics_aggs (
agent_id, bucket_seconds, bucket_start, index, name,
max_utilization, max_memory_used, max_temperature, max_power_draw, memory_total
)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
index,
MAX(name) as name,
MAX(utilization) as max_utilization,
MAX(memory_used) as max_memory_used,
MAX(temperature) as max_temperature,
MAX(power_draw) as max_power_draw,
MAX(memory_total) as memory_total
FROM gpu_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start, index
ON CONFLICT (agent_id, bucket_seconds, bucket_start, index) DO UPDATE SET
name = EXCLUDED.name,
max_utilization = EXCLUDED.max_utilization,
max_memory_used = EXCLUDED.max_memory_used,
max_temperature = EXCLUDED.max_temperature,
max_power_draw = EXCLUDED.max_power_draw,
memory_total = EXCLUDED.memory_total
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// AggregateTemperatureToAgg 将原始温度数据聚合到聚合表
func (r *MetricRepo) AggregateTemperatureToAgg(ctx context.Context, bucketSeconds int, start, end int64) error {
bucketMs := int64(bucketSeconds * 1000)
return r.db.WithContext(ctx).Exec(`
INSERT INTO temperature_metrics_aggs (
agent_id, bucket_seconds, bucket_start, sensor_key, sensor_label, max_temperature
)
SELECT
agent_id,
? as bucket_seconds,
(timestamp / ?) * ? as bucket_start,
sensor_key,
MAX(sensor_label) as sensor_label,
MAX(temperature) as max_temperature
FROM temperature_metrics
WHERE timestamp >= ? AND timestamp < ?
GROUP BY agent_id, bucket_start, sensor_key
ON CONFLICT (agent_id, bucket_seconds, bucket_start, sensor_key) DO UPDATE SET
sensor_label = EXCLUDED.sensor_label,
max_temperature = EXCLUDED.max_temperature
`, bucketSeconds, bucketMs, bucketMs, start, end).Error
}
// GetCPUMetricsAgg 从聚合表获取CPU指标
func (r *MetricRepo) GetCPUMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedCPUMetric, error) {
var metrics []AggregatedCPUMetric
err := r.db.WithContext(ctx).
Table("cpu_metrics_aggs").
Select("bucket_start as timestamp, max_usage, logical_cores").
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start").
Scan(&metrics).Error
return metrics, err
}
// GetMemoryMetricsAgg 从聚合表获取内存指标
func (r *MetricRepo) GetMemoryMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedMemoryMetric, error) {
var metrics []AggregatedMemoryMetric
err := r.db.WithContext(ctx).
Table("memory_metrics_aggs").
Select("bucket_start as timestamp, max_usage, total").
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start").
Scan(&metrics).Error
return metrics, err
}
// GetDiskMetricsAgg 从聚合表获取磁盘指标
func (r *MetricRepo) GetDiskMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedDiskMetric, error) {
var metrics []AggregatedDiskMetric
err := r.db.WithContext(ctx).
Table("disk_metrics_aggs").
Select("bucket_start as timestamp, mount_point, max_usage, total").
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start, mount_point").
Scan(&metrics).Error
return metrics, err
}
// GetNetworkMetricsAgg 从聚合表获取网络指标
func (r *MetricRepo) GetNetworkMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedNetworkMetric, error) {
var metrics []AggregatedNetworkMetric
err := r.db.WithContext(ctx).
Table("network_metrics_aggs").
Select("bucket_start as timestamp, interface, max_sent_rate, max_recv_rate").
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start, interface").
Scan(&metrics).Error
return metrics, err
}
// GetNetworkConnectionMetricsAgg 从聚合表获取网络连接指标
func (r *MetricRepo) GetNetworkConnectionMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedNetworkConnectionMetric, error) {
var metrics []AggregatedNetworkConnectionMetric
err := r.db.WithContext(ctx).
Table("network_connection_metrics_aggs").
Select(`bucket_start as timestamp,
max_established, max_syn_sent, max_syn_recv,
max_fin_wait1, max_fin_wait2, max_time_wait,
max_close, max_close_wait, max_last_ack,
max_listen, max_closing, max_total`).
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start").
Scan(&metrics).Error
return metrics, err
}
// GetDiskIOMetricsAgg 从聚合表获取磁盘IO指标
func (r *MetricRepo) GetDiskIOMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedDiskIOMetric, error) {
var metrics []AggregatedDiskIOMetric
err := r.db.WithContext(ctx).
Table("disk_io_metrics_aggs").
Select(`bucket_start as timestamp, device,
max_read_bytes_rate as max_read_rate,
max_write_bytes_rate as max_write_rate,
max_iops_in_progress`).
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start, device").
Scan(&metrics).Error
return metrics, err
}
// GetGPUMetricsAgg 从聚合表获取GPU指标
func (r *MetricRepo) GetGPUMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedGPUMetric, error) {
var metrics []AggregatedGPUMetric
err := r.db.WithContext(ctx).
Table("gpu_metrics_aggs").
Select(`bucket_start as timestamp, index, name,
max_utilization,
max_memory_used,
max_temperature,
max_power_draw,
memory_total`).
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start, index").
Scan(&metrics).Error
return metrics, err
}
// GetTemperatureMetricsAgg 从聚合表获取温度指标
func (r *MetricRepo) GetTemperatureMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedTemperatureMetric, error) {
var metrics []AggregatedTemperatureMetric
err := r.db.WithContext(ctx).
Table("temperature_metrics_aggs").
Select(`bucket_start as timestamp, sensor_key, sensor_label,
max_temperature`).
Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end).
Order("bucket_start, sensor_key").
Scan(&metrics).Error
return metrics, err
}
// UpsertAggregationProgress 更新或插入聚合进度
func (r *MetricRepo) UpsertAggregationProgress(ctx context.Context, metricType string, bucketSeconds int, lastBucket int64) error {
return r.db.WithContext(ctx).
Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "metric_type"}, {Name: "bucket_seconds"}},
DoUpdates: clause.AssignmentColumns([]string{"last_bucket"}),
}).
Create(&models.AggregationProgress{
MetricType: metricType,
BucketSeconds: bucketSeconds,
LastBucket: lastBucket,
}).Error
}
// GetAggregationProgress 获取聚合进度
func (r *MetricRepo) GetAggregationProgress(ctx context.Context, metricType string, bucketSeconds int) (*models.AggregationProgress, error) {
var progress models.AggregationProgress
err := r.db.WithContext(ctx).
Where("metric_type = ? AND bucket_seconds = ?", metricType, bucketSeconds).
First(&progress).Error
if err != nil {
return nil, err
}
return &progress, nil
}
internal/service/agent_service.go
+290 -27
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"math"
"time"
"github.com/dushixiang/pika/internal/models"
@@ -15,6 +16,19 @@ import (
"gorm.io/gorm"
)
const (
defaultMetricsRetentionHours = 24 * 7 // 默认保留 7 天
defaultMaxQueryPoints = 720 // 默认最多返回 720 个点
)
var allowedIntervals = []int{
1, 2, 5,
10, 15, 30,
60, 120, 300,
600, 900, 1800,
3600, 7200, 14400,
}
type AgentService struct {
logger *zap.Logger
*orz.Service
@@ -22,9 +36,10 @@ type AgentService struct {
metricRepo *repo.MetricRepo
monitorStatsRepo *repo.MonitorStatsRepo
apiKeyService *ApiKeyService
propertyService *PropertyService
}
func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService) *AgentService {
func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService, propertyService *PropertyService) *AgentService {
return &AgentService{
logger: logger,
Service: orz.NewService(db),
@@ -32,6 +47,7 @@ func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyServi
metricRepo: repo.NewMetricRepo(db),
monitorStatsRepo: repo.NewMonitorStatsRepo(db),
apiKeyService: apiKeyService,
propertyService: propertyService,
}
}
@@ -246,20 +262,6 @@ func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, met
}
return s.metricRepo.SaveNetworkConnectionMetric(ctx, metric)
case protocol.MetricTypeLoad:
var loadData protocol.LoadData
if err := json.Unmarshal(data, &loadData); err != nil {
return err
}
metric := &models.LoadMetric{
AgentID: agentID,
Load1: loadData.Load1,
Load5: loadData.Load5,
Load15: loadData.Load15,
Timestamp: now,
}
return s.metricRepo.SaveLoadMetric(ctx, metric)
case protocol.MetricTypeDiskIO:
// DiskIO现在是数组,需要批量处理
var diskIODataList []*protocol.DiskIOData
@@ -424,26 +426,96 @@ func CalculateInterval(start, end int64) int {
}
}
// GetMetrics 获取聚合指标数据
// GetMetrics 获取聚合指标数据（自动路由到聚合表或原始表）
func (s *AgentService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interval int) (interface{}, error) {
start, end = s.normalizeTimeRange(ctx, start, end)
interval = s.DetermineInterval(ctx, start, end, interval)
// 判断是否可以使用聚合表（仅支持部分指标类型）
aggCapable := map[string]bool{
"cpu": true,
"memory": true,
"disk": true,
"network": true,
"network_connection": true,
"disk_io": true,
"gpu": true,
"temperature": true,
}
// 智能选择聚合粒度：根据查询间隔选择最合适的bucket
// 例如：查询90秒数据时使用60秒bucket，查询600秒数据时使用300秒bucket
var bucketSeconds int
useAgg := false
if aggCapable[metricType] {
if interval >= 3600 {
bucketSeconds = 3600
useAgg = true
} else if interval >= 300 {
bucketSeconds = 300
useAgg = true
} else if interval >= 60 {
bucketSeconds = 60
useAgg = true
}
}
switch metricType {
case "cpu":
if useAgg {
if metrics, err := s.metricRepo.GetCPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetCPUMetrics(ctx, agentID, start, end, interval)
case "memory":
if useAgg {
if metrics, err := s.metricRepo.GetMemoryMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetMemoryMetrics(ctx, agentID, start, end, interval)
case "disk":
if useAgg {
if metrics, err := s.metricRepo.GetDiskMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetDiskMetrics(ctx, agentID, start, end, interval)
case "network":
if useAgg {
if metrics, err := s.metricRepo.GetNetworkMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetNetworkMetrics(ctx, agentID, start, end, interval)
case "network_connection":
if useAgg {
if metrics, err := s.metricRepo.GetNetworkConnectionMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetNetworkConnectionMetrics(ctx, agentID, start, end, interval)
case "load":
return s.metricRepo.GetLoadMetrics(ctx, agentID, start, end, interval)
case "disk_io":
if useAgg {
if metrics, err := s.metricRepo.GetDiskIOMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetDiskIOMetrics(ctx, agentID, start, end, interval)
case "gpu":
if useAgg {
if metrics, err := s.metricRepo.GetGPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetGPUMetrics(ctx, agentID, start, end, interval)
case "temperature":
if useAgg {
if metrics, err := s.metricRepo.GetTemperatureMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 {
return metrics, nil
}
}
return s.metricRepo.GetTemperatureMetrics(ctx, agentID, start, end, interval)
default:
return nil, nil
@@ -452,9 +524,187 @@ func (s *AgentService) GetMetrics(ctx context.Context, agentID, metricType strin
// GetNetworkMetricsByInterface 获取按网卡接口分组的网络指标
func (s *AgentService) GetNetworkMetricsByInterface(ctx context.Context, agentID string, start, end int64, interval int) (interface{}, error) {
start, end = s.normalizeTimeRange(ctx, start, end)
interval = s.DetermineInterval(ctx, start, end, interval)
return s.metricRepo.GetNetworkMetricsByInterface(ctx, agentID, start, end, interval)
}
// DetermineInterval 根据配置、用户请求和时间范围决定聚合粒度
func (s *AgentService) DetermineInterval(ctx context.Context, start, end int64, requested int) int {
cfg := s.getMetricsConfig(ctx)
interval := requested
if interval <= 0 {
interval = calculateBaseInterval(start, end)
}
interval = adjustIntervalForMaxPoints(start, end, interval, cfg.MaxQueryPoints)
return interval
}
// normalizeTimeRange 将时间范围限制在保留周期内，避免无意义的全表扫描
func (s *AgentService) normalizeTimeRange(ctx context.Context, start, end int64) (int64, int64) {
cfg := s.getMetricsConfig(ctx)
retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour
retentionBoundary := time.Now().Add(-retentionDuration).UnixMilli()
if start < retentionBoundary {
start = retentionBoundary
}
if end <= start {
end = start + 1000
}
return start, end
}
// 默认间隔选择（保留旧行为的同时限制最大数据点）
func defaultIntervalWithLimit(start, end int64) int {
base := calculateBaseInterval(start, end)
return adjustIntervalForMaxPoints(start, end, base, defaultMaxQueryPoints)
}
func calculateBaseInterval(start, end int64) int {
duration := (end - start) / 1000 // 秒
switch {
case duration <= 60: // 1 分钟内
return 2
case duration <= 5*60: // 5 分钟内
return 5
case duration <= 15*60: // 15 分钟内
return 15
case duration <= 30*60: // 30 分钟内
return 30
case duration <= 60*60: // 1 小时内
return 60
case duration <= 3*60*60: // 3 小时内
return 180
case duration <= 6*60*60: // 6 小时内
return 300
case duration <= 12*60*60: // 12 小时内
return 600
case duration <= 24*60*60: // 1 天内
return 900
case duration <= 3*24*60*60: // 3 天内
return 1800
case duration <= 7*24*60*60: // 7 天内
return 3600
case duration <= 14*24*60*60: // 14 天内
return 7200
default:
return 14400 // 更长时间：4 小时粒度
}
}
// adjustIntervalForMaxPoints 根据最大数据点限制提升聚合粒度
func adjustIntervalForMaxPoints(start, end int64, interval int, maxPoints int) int {
if interval <= 0 {
interval = 1
}
if maxPoints <= 0 {
return alignInterval(interval)
}
durationSeconds := float64(end-start) / 1000
if durationSeconds <= 0 {
return alignInterval(interval)
}
required := int(math.Ceil(durationSeconds / float64(maxPoints)))
interval = maxInt(interval, required)
return alignInterval(interval)
}
func alignInterval(interval int) int {
for _, candidate := range allowedIntervals {
if interval <= candidate {
return candidate
}
}
return allowedIntervals[len(allowedIntervals)-1]
}
func maxInt(a, b int) int {
if a > b {
return a
}
return b
}
func (s *AgentService) getMetricsConfig(ctx context.Context) models.MetricsConfig {
cfg := models.MetricsConfig{
RetentionHours: defaultMetricsRetentionHours,
MaxQueryPoints: defaultMaxQueryPoints,
}
if s.propertyService == nil {
return cfg
}
loaded := s.propertyService.GetMetricsConfig(ctx)
if loaded.RetentionHours > 0 {
cfg.RetentionHours = loaded.RetentionHours
}
if loaded.MaxQueryPoints > 0 {
cfg.MaxQueryPoints = loaded.MaxQueryPoints
}
return cfg
}
// runAggregation 按固定 bucket 下采样存储
func (s *AgentService) runAggregation(ctx context.Context) {
cfg := s.getMetricsConfig(ctx)
retention := time.Duration(cfg.RetentionHours) * time.Hour
// 标准 bucket：1m、5m、1h
buckets := []int{60, 300, 3600}
for _, bucket := range buckets {
s.aggregateMetric(ctx, "cpu", bucket, retention, s.metricRepo.AggregateCPUToAgg)
s.aggregateMetric(ctx, "memory", bucket, retention, s.metricRepo.AggregateMemoryToAgg)
s.aggregateMetric(ctx, "disk", bucket, retention, s.metricRepo.AggregateDiskToAgg)
s.aggregateMetric(ctx, "network", bucket, retention, s.metricRepo.AggregateNetworkToAgg)
s.aggregateMetric(ctx, "network_connection", bucket, retention, s.metricRepo.AggregateNetworkConnectionToAgg)
s.aggregateMetric(ctx, "disk_io", bucket, retention, s.metricRepo.AggregateDiskIOToAgg)
s.aggregateMetric(ctx, "gpu", bucket, retention, s.metricRepo.AggregateGPUToAgg)
s.aggregateMetric(ctx, "temperature", bucket, retention, s.metricRepo.AggregateTemperatureToAgg)
}
}
type aggregateFn func(ctx context.Context, bucketSeconds int, start, end int64) error
func (s *AgentService) aggregateMetric(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, fn aggregateFn) {
bucketMs := int64(bucketSeconds * 1000)
start := s.getAggregationStart(ctx, metricType, bucketSeconds, retention, bucketMs)
endBucket := (time.Now().Add(-time.Duration(bucketSeconds)*time.Second).UnixMilli() / bucketMs) * bucketMs
if endBucket <= start {
return
}
end := endBucket + bucketMs - 1
if err := fn(ctx, bucketSeconds, start, end); err != nil {
s.logger.Error("aggregate metric failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err))
return
}
if err := s.metricRepo.UpsertAggregationProgress(ctx, metricType, bucketSeconds, endBucket); err != nil {
s.logger.Error("update aggregation progress failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err))
}
}
func (s *AgentService) getAggregationStart(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, bucketMs int64) int64 {
progress, err := s.metricRepo.GetAggregationProgress(ctx, metricType, bucketSeconds)
if err == nil && progress != nil && progress.LastBucket > 0 {
return progress.LastBucket + bucketMs
}
// 默认从保留窗口开始
start := time.Now().Add(-retention).UnixMilli()
return (start / bucketMs) * bucketMs
}
// GetLatestMetrics 获取最新指标
func (s *AgentService) GetLatestMetrics(ctx context.Context, agentID string) (*LatestMetrics, error) {
result := &LatestMetrics{}
@@ -507,11 +757,6 @@ func (s *AgentService) GetLatestMetrics(ctx context.Context, agentID string) (*L
}
}
// 获取最新负载信息
if load, err := s.metricRepo.GetLatestLoadMetric(ctx, agentID); err == nil {
result.Load = load
}
// 获取最新主机信息
if host, err := s.metricRepo.GetLatestHostMetric(ctx, agentID); err == nil {
result.Host = host
@@ -555,10 +800,11 @@ func (s *AgentService) StartCleanupTask(ctx context.Context) {
// cleanupOldMetrics 清理旧数据
func (s *AgentService) cleanupOldMetrics(ctx context.Context) {
// 删除1小时前的数据
before := time.Now().Add(-1 * time.Hour).UnixMilli()
cfg := s.getMetricsConfig(ctx)
retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour
before := time.Now().Add(-retentionDuration).UnixMilli()
s.logger.Info("starting to clean old metrics", zap.Int64("beforeTimestamp", before))
s.logger.Info("starting to clean old metrics", zap.Int64("beforeTimestamp", before), zap.Int("retentionHours", cfg.RetentionHours))
if err := s.metricRepo.DeleteOldMetrics(ctx, before); err != nil {
s.logger.Error("failed to clean old metrics", zap.Error(err))
@@ -568,6 +814,24 @@ func (s *AgentService) cleanupOldMetrics(ctx context.Context) {
s.logger.Info("old metrics cleaned successfully")
}
// StartAggregationTask 启动聚合下采样任务
func (s *AgentService) StartAggregationTask(ctx context.Context) {
ticker := time.NewTicker(1 * time.Minute)
defer ticker.Stop()
s.logger.Info("aggregation task started")
for {
select {
case <-ctx.Done():
s.logger.Info("aggregation task stopped")
return
case <-ticker.C:
s.runAggregation(ctx)
}
}
}
// DiskSummary 磁盘汇总数据
type DiskSummary struct {
AvgUsagePercent float64 `json:"avgUsagePercent"` // 平均使用率
@@ -593,7 +857,6 @@ type LatestMetrics struct {
Disk *DiskSummary `json:"disk,omitempty"`
Network *NetworkSummary `json:"network,omitempty"`
NetworkConnection *models.NetworkConnectionMetric `json:"networkConnection,omitempty"`
Load *models.LoadMetric `json:"load,omitempty"`
Host *models.HostMetric `json:"host,omitempty"`
GPU []models.GPUMetric `json:"gpu,omitempty"`
Temp []models.TemperatureMetric `json:"temperature,omitempty"`
internal/service/property_service.go
+26
View File
@@ -19,6 +19,8 @@ const (
PropertyIDNotificationChannels = "notification_channels"
// PropertyIDSystemConfig 系统配置的固定 ID
PropertyIDSystemConfig = "system_config"
// PropertyIDMetricsConfig 指标配置的固定 ID
PropertyIDMetricsConfig = "metrics_config"
)
type PropertyService struct {
@@ -123,6 +125,22 @@ func (s *PropertyService) GetSystemConfig(ctx context.Context) (*models.SystemCo
return &systemConfig, nil
}
// GetMetricsConfig 获取指标配置
func (s *PropertyService) GetMetricsConfig(ctx context.Context) models.MetricsConfig {
var config models.MetricsConfig
err := s.GetValue(ctx, PropertyIDMetricsConfig, &config)
if err != nil {
// 返回默认配置
return models.MetricsConfig{}
}
return config
}
// SetMetricsConfig 设置指标配置
func (s *PropertyService) SetMetricsConfig(ctx context.Context, config models.MetricsConfig) error {
return s.Set(ctx, PropertyIDMetricsConfig, "指标数据配置", config)
}
// defaultPropertyConfig 默认配置项定义
type defaultPropertyConfig struct {
ID string
@@ -150,6 +168,14 @@ func (s *PropertyService) InitializeDefaultConfigs(ctx context.Context) error {
Name: "通知渠道配置",
Value: []models.NotificationChannelConfig{},
},
{
ID: PropertyIDMetricsConfig,
Name: "指标数据配置",
Value: models.MetricsConfig{
RetentionHours: 168, // 默认7天
MaxQueryPoints: 720, // 默认720个点
},
},
}
// 遍历并初始化每个配置
internal/wire_gen.go
+2 -2
View File
@@ -26,14 +26,14 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App
accountService := service.NewAccountService(logger, userService, oidcService, gitHubOAuthService, cfg)
accountHandler := handler.NewAccountHandler(accountService)
apiKeyService := service.NewApiKeyService(logger, db)
agentService := service.NewAgentService(logger, db, apiKeyService)
propertyService := service.NewPropertyService(logger, db)
agentService := service.NewAgentService(logger, db, apiKeyService, propertyService)
manager := websocket.NewManager(logger)
monitorService := service.NewMonitorService(logger, db, manager)
tamperRepo := repo.NewTamperRepo(db)
tamperService := service.NewTamperService(logger, tamperRepo, manager)
agentHandler := handler.NewAgentHandler(logger, agentService, monitorService, tamperService, manager)
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)
pkg/agent/collector/manager.go
-11
View File
@@ -20,7 +20,6 @@ type Manager struct {
diskIOCollector *DiskIOCollector
networkCollector *NetworkCollector
networkConnectionCollector *NetworkConnectionCollector
loadCollector *LoadCollector
hostCollector *HostCollector
temperatureCollector *TemperatureCollector
gpuCollector *GPUCollector
@@ -36,7 +35,6 @@ func NewManager(cfg *config.Config) *Manager {
diskIOCollector: NewDiskIOCollector(),
networkCollector: NewNetworkCollector(cfg),
networkConnectionCollector: NewNetworkConnectionCollector(),
loadCollector: NewLoadCollector(),
hostCollector: NewHostCollector(),
temperatureCollector: NewTemperatureCollector(),
gpuCollector: NewGPUCollector(),
@@ -100,15 +98,6 @@ func (m *Manager) CollectAndSendNetworkConnection(conn WebSocketWriter) error {
return m.sendMetrics(conn, protocol.MetricTypeNetworkConnection, connectionData)
}
// CollectAndSendLoad 采集并发送系统负载指标
func (m *Manager) CollectAndSendLoad(conn WebSocketWriter) error {
loadData, err := m.loadCollector.Collect()
if err != nil {
return err
}
return m.sendMetrics(conn, protocol.MetricTypeLoad, loadData)
}
// CollectAndSendHost 采集并发送主机信息
func (m *Manager) CollectAndSendHost(conn WebSocketWriter) error {
hostData, err := m.hostCollector.Collect()
pkg/agent/collector/system.go
-23
View File
@@ -3,31 +3,8 @@ package collector
import (
"github.com/dushixiang/pika/internal/protocol"
"github.com/shirou/gopsutil/v4/host"
"github.com/shirou/gopsutil/v4/load"
)
// LoadCollector 系统负载监控采集器
type LoadCollector struct{}
// NewLoadCollector 创建负载采集器
func NewLoadCollector() *LoadCollector {
return &LoadCollector{}
}
// Collect 采集系统负载数据
func (l *LoadCollector) Collect() (*protocol.LoadData, error) {
avgStat, err := load.Avg()
if err != nil {
return nil, err
}
return &protocol.LoadData{
Load1: avgStat.Load1,
Load5: avgStat.Load5,
Load15: avgStat.Load15,
}, nil
}
// HostCollector 主机信息采集器
type HostCollector struct{}
pkg/agent/service/agent.go
-6
View File
@@ -482,12 +482,6 @@ func (a *Agent) collectAndSendAllMetrics(conn *safeConn, manager *collector.Mana
hasError = true
}
// 系统负载指标
if err := manager.CollectAndSendLoad(conn); err != nil {
log.Printf("⚠️ 发送负载指标失败: %v", err)
hasError = true
}
// 主机信息
if err := manager.CollectAndSendHost(conn); err != nil {
log.Printf("⚠️ 发送主机信息失败: %v", err)
web/src/api/agent.ts
+3 -3
View File
@@ -8,8 +8,8 @@ export interface ListAgentsResponse {
export interface GetAgentMetricsRequest {
agentId: string;
type: 'cpu' | 'memory' | 'disk' | 'network' | 'network_connection' | 'load' | 'disk_io' | 'gpu' | 'temperature';
range?: '1m' | '5m' | '15m' | '30m' | '1h';
type: 'cpu' | 'memory' | 'disk' | 'network' | 'network_connection' | 'disk_io' | 'gpu' | 'temperature';
range?: '1m' | '5m' | '15m' | '30m' | '1h' | '3h' | '6h' | '12h' | '1d' | '24h' | '3d' | '7d' | '30d';
}
export interface GetAgentMetricsResponse {
@@ -74,7 +74,7 @@ export const getAgentLatestMetrics = (agentId: string) => {
export interface GetNetworkMetricsByInterfaceRequest {
agentId: string;
range?: '1m' | '5m' | '15m' | '30m' | '1h';
range?: '1m' | '5m' | '15m' | '30m' | '1h' | '3h' | '6h' | '12h' | '1d' | '24h' | '3d' | '7d' | '30d';
}
export interface NetworkMetricByInterface {
web/src/pages/Public/ServerDetail.tsx
+17 -11
View File
@@ -94,6 +94,12 @@ const timeRangeOptions = [
{label: '15分钟', value: '15m'},
{label: '30分钟', value: '30m'},
{label: '1小时', value: '1h'},
{label: '3小时', value: '3h'},
{label: '6小时', value: '6h'},
{label: '12小时', value: '12h'},
{label: '1天', value: '1d'},
{label: '3天', value: '3d'},
{label: '7天', value: '7d'},
] as const;
type TimeRange = typeof timeRangeOptions[number]['value'];
@@ -230,9 +236,9 @@ const TimeRangeSelector = ({
key={option.value}
type="button"
onClick={() => onChange(option.value)}
className={`rounded-lg border px-3 py-1.5 text-sm transition ${
className={`rounded-lg border px-3 py-1.5 text-xs sm:text-sm font-medium transition whitespace-nowrap ${
isActive
? 'border-blue-200 dark:border-blue-400 bg-blue-600 dark:bg-blue-500 text-white'
? 'border-blue-200 dark:border-blue-400 bg-blue-600 dark:bg-blue-500 text-white shadow-sm'
: 'border-slate-200 dark:border-slate-800 bg-white dark:bg-slate-900 text-slate-600 dark:text-slate-300 hover:border-blue-200 dark:hover:border-blue-500 hover:text-blue-600 dark:hover:text-blue-200'
}`}
>
@@ -744,19 +750,19 @@ const ServerDetail = () => {
});
cards.push({
key: 'load',
icon: Activity,
title: '系统负载',
usagePercent: latestMetrics.load
? `${latestMetrics.load.load1.toFixed(2)}`
key: 'network',
icon: Network,
title: '网络流量',
usagePercent: latestMetrics.network
? `${formatBytes(latestMetrics.network.totalBytesSentRate)}/s`
: '—',
accent: 'amber',
metrics: [
{
label: '1 / 5 / 15 分钟',
value: latestMetrics.load
? `${latestMetrics.load.load1.toFixed(2)} / ${latestMetrics.load.load5.toFixed(2)} / ${latestMetrics.load.load15.toFixed(2)}`
: '-',
label: '上行 / 下行',
value: `${formatBytes(latestMetrics.network?.totalBytesSentRate)}/s ↑ / ${formatBytes(
latestMetrics.network?.totalBytesRecvRate,
)}/s ↓`,
},
{
label: '网络累计',
web/src/pages/Settings/MetricsConfig.tsx
+288
View File
@@ -0,0 +1,288 @@
import {useEffect} from 'react';
import {App, Button, Card, Form, InputNumber, Space, Spin} from 'antd';
import {Database, Clock, BarChart3} from 'lucide-react';
import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query';
import type {MetricsConfig} from '../../api/metrics-config';
import {getMetricsConfig, saveMetricsConfig} from '../../api/metrics-config';
import {getErrorMessage} from '../../lib/utils';
const MetricsConfigComponent = () => {
const [form] = Form.useForm();
const {message: messageApi} = App.useApp();
const queryClient = useQueryClient();
// 获取指标配置
const {data: metricsConfig, isLoading} = useQuery({
queryKey: ['metricsConfig'],
queryFn: getMetricsConfig,
});
// 保存指标配置 mutation
const saveMutation = useMutation({
mutationFn: saveMetricsConfig,
onSuccess: () => {
messageApi.success('配置保存成功');
queryClient.invalidateQueries({queryKey: ['metricsConfig']});
},
onError: (error: unknown) => {
messageApi.error(getErrorMessage(error, '保存配置失败'));
},
});
// 初始化表单
useEffect(() => {
if (metricsConfig) {
form.setFieldsValue({
retentionHours: metricsConfig.retentionHours,
maxQueryPoints: metricsConfig.maxQueryPoints,
});
}
}, [metricsConfig, form]);
const handleSave = async () => {
try {
const values = await form.validateFields();
saveMutation.mutate({
retentionHours: values.retentionHours,
maxQueryPoints: values.maxQueryPoints,
} as MetricsConfig);
} catch (error) {
// 表单验证失败
}
};
const handleReset = () => {
if (metricsConfig) {
form.setFieldsValue({
retentionHours: metricsConfig.retentionHours,
maxQueryPoints: metricsConfig.maxQueryPoints,
});
}
};
// 计算保留天数
const getRetentionDays = (hours: number) => {
return (hours / 24).toFixed(1);
};
if (isLoading) {
return (
<div className="flex justify-center items-center py-20">
<Spin size="large"/>
</div>
);
}
return (
<div>
<div className="mb-6">
<h2 className="text-xl font-bold flex items-center gap-2">
<Database size={20}/>
指标数据配置
</h2>
<p className="text-gray-500 mt-2">配置监控指标数据的保留时长和查询限制，优化存储空间和查询性能</p>
</div>
<Form form={form} layout="vertical" onFinish={handleSave}>
<Space direction="vertical" className="w-full" size="large">
{/* 数据保留策略 */}
<Card
title={
<div className="flex items-center gap-2">
<Clock size={18}/>
<span>数据保留策略</span>
</div>
}
type="inner"
>
<Form.Item
label="数据保留时长"
name="retentionHours"
rules={[
{required: true, message: '请输入数据保留时长'},
{
type: 'number',
min: 24,
max: 720,
message: '保留时长必须在 24-720 小时之间（1-30天）'
},
]}
tooltip="原始指标数据和聚合数据的保留时长，超过此时长的数据将被自动清理"
>
<InputNumber
min={24}
max={720}
step={24}
addonAfter="小时"
style={{width: 200}}
placeholder="168"
/>
</Form.Item>
<Form.Item noStyle shouldUpdate>
{({getFieldValue}) => {
const hours = getFieldValue('retentionHours');
return hours ? (
<div className="mb-4 text-sm text-gray-500">
当前设置约为 <span
className="font-semibold text-blue-600">{getRetentionDays(hours)}</span> 天
</div>
) : null;
}}
</Form.Item>
<div className="p-4 bg-blue-50 dark:bg-blue-950/20 rounded-lg border border-blue-200 dark:border-blue-800">
<div className="text-sm text-blue-800 dark:text-blue-300 space-y-2">
<div className="font-semibold flex items-center gap-2">
💡 保留策略说明
</div>
<ul className="list-disc list-inside space-y-1.5 ml-2">
<li>系统会保留<strong>原始数据</strong>和<strong>3种粒度的聚合数据</strong>（1分钟、5分钟、1小时）</li>
<li>聚合数据自动生成，用于优化长时间范围的查询性能</li>
<li>较短的保留时长可以<strong>节省存储空间</strong>，减少数据库压力</li>
<li>修改后立即生效，下次清理任务时应用新策略（每小时执行一次）</li>
</ul>
</div>
</div>
<div className="mt-4 p-4 bg-slate-50 dark:bg-slate-800/50 rounded-lg">
<div className="text-sm space-y-2">
<div className="font-semibold text-slate-700 dark:text-slate-300">📊 推荐设置：</div>
<div className="grid grid-cols-1 sm:grid-cols-3 gap-3">
<div className="p-3 bg-white dark:bg-slate-900 rounded border border-slate-200 dark:border-slate-700">
<div className="font-medium text-slate-900 dark:text-slate-100">7 天（168小时）</div>
<div className="text-xs text-slate-500 dark:text-slate-400 mt-1">适合日常监控</div>
</div>
<div className="p-3 bg-white dark:bg-slate-900 rounded border border-slate-200 dark:border-slate-700">
<div className="font-medium text-slate-900 dark:text-slate-100">14 天（336小时）</div>
<div className="text-xs text-slate-500 dark:text-slate-400 mt-1">适合周期分析</div>
</div>
<div className="p-3 bg-white dark:bg-slate-900 rounded border border-slate-200 dark:border-slate-700">
<div className="font-medium text-slate-900 dark:text-slate-100">30 天（720小时）</div>
<div className="text-xs text-slate-500 dark:text-slate-400 mt-1">适合长期趋势</div>
</div>
</div>
</div>
</div>
</Card>
{/* 查询性能优化 */}
<Card
title={
<div className="flex items-center gap-2">
<BarChart3 size={18}/>
<span>查询性能优化</span>
</div>
}
type="inner"
>
<Form.Item
label="最大查询点数"
name="maxQueryPoints"
rules={[
{required: true, message: '请输入最大查询点数'},
{
type: 'number',
min: 100,
max: 10000,
message: '查询点数必须在 100-10000 之间'
},
]}
tooltip="单次查询返回的最大数据点数，系统会根据此值自动选择合适的聚合粒度"
>
<InputNumber
min={100}
max={10000}
step={100}
addonAfter="个数据点"
style={{width: 200}}
placeholder="720"
/>
</Form.Item>
<div className="p-4 bg-emerald-50 dark:bg-emerald-950/20 rounded-lg border border-emerald-200 dark:border-emerald-800">
<div className="text-sm text-emerald-800 dark:text-emerald-300 space-y-2">
<div className="font-semibold flex items-center gap-2">
✨ 智能聚合机制
</div>
<ul className="list-disc list-inside space-y-1.5 ml-2">
<li>查询时间范围越长，系统会自动使用<strong>更大粒度的聚合数据</strong></li>
<li>确保返回的数据点数不超过设定的最大值，<strong>优化传输和渲染</strong></li>
<li>例如：查询7天数据时自动使用1小时聚合，而不是原始秒级数据</li>
<li>聚合使用<strong>MAX（最大值）</strong>策略，确保不会遗漏峰值</li>
</ul>
</div>
</div>
<div className="mt-4 p-4 bg-amber-50 dark:bg-amber-950/20 rounded-lg border border-amber-200 dark:border-amber-800">
<div className="text-sm text-amber-800 dark:text-amber-300 space-y-2">
<div className="font-semibold flex items-center gap-2">
⚠️ 性能考量
</div>
<ul className="list-disc list-inside space-y-1.5 ml-2">
<li>较大的点数可以提供<strong>更精细的图表</strong>，但会增加<strong>查询时间和带宽消耗</strong></li>
<li>建议值：<strong>720 个点</strong>（适合显示24小时数据，每2分钟一个点）</li>
<li>前端图表通常显示 <strong>300-1000 个点</strong>效果最佳，过多的点反而影响渲染性能</li>
<li>移动端建议使用<strong>较小的值</strong>（300-500），以节省流量和提升加载速度</li>
</ul>
</div>
</div>
</Card>
{/* 聚合数据说明 */}
<Card
title="聚合粒度说明"
type="inner"
className="bg-slate-50 dark:bg-slate-900/50"
>
<div className="text-sm space-y-3">
<p className="text-slate-600 dark:text-slate-400">
系统会在后台自动将原始数据聚合为不同粒度的数据，以支持不同时间范围的查询：
</p>
<div className="grid grid-cols-1 md:grid-cols-3 gap-4">
<div className="p-4 bg-white dark:bg-slate-800 rounded-lg border border-slate-200 dark:border-slate-700">
<div className="flex items-center gap-2 mb-2">
<div className="w-2 h-2 rounded-full bg-blue-500"></div>
<span className="font-semibold text-slate-900 dark:text-slate-100">1 分钟聚合</span>
</div>
<p className="text-xs text-slate-600 dark:text-slate-400">适合查询 15分钟 - 1小时 的数据</p>
</div>
<div className="p-4 bg-white dark:bg-slate-800 rounded-lg border border-slate-200 dark:border-slate-700">
<div className="flex items-center gap-2 mb-2">
<div className="w-2 h-2 rounded-full bg-emerald-500"></div>
<span className="font-semibold text-slate-900 dark:text-slate-100">5 分钟聚合</span>
</div>
<p className="text-xs text-slate-600 dark:text-slate-400">适合查询 1小时 - 12小时 的数据</p>
</div>
<div className="p-4 bg-white dark:bg-slate-800 rounded-lg border border-slate-200 dark:border-slate-700">
<div className="flex items-center gap-2 mb-2">
<div className="w-2 h-2 rounded-full bg-purple-500"></div>
<span className="font-semibold text-slate-900 dark:text-slate-100">1 小时聚合</span>
</div>
<p className="text-xs text-slate-600 dark:text-slate-400">适合查询 12小时 - 7天 的数据</p>
</div>
</div>
<p className="text-xs text-slate-500 dark:text-slate-400 mt-3">
💡 提示：系统会自动选择最合适的聚合粒度，无需手动配置
</p>
</div>
</Card>
{/* 保存按钮 */}
<Form.Item>
<Space>
<Button type="primary" htmlType="submit" loading={saveMutation.isPending} size="large">
保存配置
</Button>
<Button onClick={handleReset} size="large">
重置
</Button>
</Space>
</Form.Item>
</Space>
</Form>
</div>
);
};
export default MetricsConfigComponent;
web/src/pages/Settings/SystemConfig.tsx
+2 -2
View File
@@ -20,7 +20,7 @@ const SystemConfigComponent = () => {
queryFn: getSystemConfig,
});
// 保存 mutation
// 保存系统配置 mutation
const saveMutation = useMutation({
mutationFn: saveSystemConfig,
onSuccess: () => {
@@ -34,7 +34,7 @@ const SystemConfigComponent = () => {
},
});
// 初始化表单值
// 初始化系统配置表单
useEffect(() => {
if (config) {
form.setFieldsValue({
web/src/pages/Settings/index.tsx
+12 -1
View File
@@ -1,8 +1,9 @@
import {Tabs} from 'antd';
import {Bell, MessageSquare, Settings2} from 'lucide-react';
import {Bell, MessageSquare, Settings2, Database} from 'lucide-react';
import AlertSettings from './AlertSettings';
import NotificationChannels from './NotificationChannels';
import SystemConfig from './SystemConfig';
import MetricsConfig from './MetricsConfig';
import {PageHeader} from "../../components";
const Settings = () => {
@@ -17,6 +18,16 @@ const Settings = () => {
),
children: <SystemConfig/>,
},
{
key: 'metrics',
label: (
<span className="flex items-center gap-2">
<Database size={16}/>
指标数据配置
</span>
),
children: <MetricsConfig/>,
},
{
key: 'channels',
label: (
web/src/types/index.ts
-19
View File
@@ -58,15 +58,6 @@ export interface AggregatedNetworkMetric {
interface: string;
maxSentRate: number;
maxRecvRate: number;
totalSent: number;
totalRecv: number;
}
export interface AggregatedLoadMetric {
timestamp: number;
maxLoad1: number;
maxLoad5: number;
maxLoad15: number;
}
export interface AggregatedDiskMetric {
@@ -140,15 +131,6 @@ export interface MemoryMetric {
swapFree: number;
}
export interface LoadMetric {
id: string;
agentId: string;
timestamp: number;
load1: number;
load5: number;
load15: number;
}
// 磁盘汇总数据
export interface DiskSummary {
avgUsagePercent: number; // 平均使用率
@@ -405,7 +387,6 @@ export interface LatestMetrics {
disk?: DiskSummary; // 改为汇总数据
network?: NetworkSummary; // 改为汇总数据
networkConnection?: NetworkConnectionMetric; // 网络连接统计
load?: LoadMetric;
host?: HostMetric; // 主机信息
gpu?: GPUMetric[]; // GPU 列表
temperature?: TemperatureMetric[]; // 温度传感器列表