fix: #626 CPU占用过大

database/metricstore/metrics.go

@@ -50,6 +50,43 @@ var recordMetricNames = joinMetricNames(loadRecordMetricNames, gpuDeviceRecordMe
 // Ping has an independent retention and cleanup boundary.
 var pingMetricNames = []string{MetricPingLatency, MetricPingLoss}
 
+func metricNameForRecordField(name string) (string, bool) {
+	switch name {
+	case "cpu":
+		return MetricCPU, true
+	case "gpu":
+		return MetricGPU, true
+	case "ram":
+		return MetricRAM, true
+	case "swap":
+		return MetricSwap, true
+	case "load":
+		return MetricLoad, true
+	case "disk":
+		return MetricDisk, true
+	case "net_in", "netin":
+		return MetricNetIn, true
+	case "net_out", "netout":
+		return MetricNetOut, true
+	case "net_total_up":
+		return MetricNetTotalUp, true
+	case "net_total_down":
+		return MetricNetTotalDown, true
+	case "traffic_up":
+		return MetricTrafficUp, true
+	case "traffic_down":
+		return MetricTrafficDown, true
+	case "process":
+		return MetricProcess, true
+	case "connections":
+		return MetricConnections, true
+	case "connections_udp":
+		return MetricConnectionsUDP, true
+	default:
+		return "", false
+	}
+}
+
 func joinMetricNames(groups ...[]string) []string {
 	total := 0
 	for _, group := range groups {

database/metricstore/metricstore.go

@@ -782,6 +782,16 @@ func GetRecordsByClientAndTime(ctx context.Context, clientUUID string, start, en
 	return getRecordsByClientAndTimeFromSeries(ctx, s, clientUUID, start, end)
 }
 
+// GetRecordMetricMaxByClientAndTime 查询单项监控指标在各时间桶内的最大值。
+func GetRecordMetricMaxByClientAndTime(ctx context.Context, clientUUID, recordMetric string, start, end time.Time) ([]models.Record, error) {
+	s := GetStore()
+	if s == nil {
+		return nil, fmt.Errorf("metric store not enabled")
+	}
+
+	return getRecordMetricMaxByClientAndTimeFromSeries(ctx, s, clientUUID, recordMetric, start, end)
+}
+
 // GetRecordsByTime 从 metric store 查询所有客户端在时间范围内的记录
 func GetRecordsByTime(ctx context.Context, start, end time.Time) ([]models.Record, error) {
 	s := GetStore()
@@ -855,6 +865,41 @@ func getRecordsByClientAndTimeFromSeries(ctx context.Context, s *metric.Store, c
 	return records, nil
 }
 
+func getRecordMetricMaxByClientAndTimeFromSeries(ctx context.Context, s *metric.Store, clientUUID, recordMetric string, start, end time.Time) ([]models.Record, error) {
+	metricName, ok := metricNameForRecordField(recordMetric)
+	if !ok {
+		return nil, fmt.Errorf("unsupported record metric %q", recordMetric)
+	}
+
+	now := time.Now().UTC()
+	points, err := s.Series(ctx, metric.AggregateQuery{
+		Query: metric.Query{
+			MetricName: metricName,
+			EntityID:   clientUUID,
+			Start:      start,
+			End:        end,
+			Order:      metric.OrderAsc,
+		},
+		Aggregation: metric.AggMax,
+		Interval:    recordSeriesInterval(s, start, end, now),
+	}, now)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query metric %s: %w", metricName, err)
+	}
+
+	records := make([]models.Record, 0, len(points))
+	for _, point := range points {
+		entityID := point.EntityID
+		if entityID == "" {
+			entityID = clientUUID
+		}
+		record := models.Record{Client: entityID, Time: point.Bucket.UTC()}
+		applyRecordMetricValue(&record, metricName, point.Value)
+		records = append(records, record)
+	}
+	return records, nil
+}
+
 func recordMetricAggregation(metricName string) metric.Aggregation {
 	switch metricName {
 	case MetricTrafficUp, MetricTrafficDown:

database/metricstore/metricstore_test.go

@@ -566,3 +566,41 @@ func TestGetRecordsByClientAndTimeReadsRollupsAfterRawCompaction(t *testing.T) {
 		t.Fatalf("all-client records were not reconstructed from rollup: %#v", all)
 	}
 }
+
+func TestGetRecordMetricMaxByClientAndTimeQueriesOnlySelectedMetric(t *testing.T) {
+	ctx := context.Background()
+	s, err := metric.Open(ctx, metric.SQLite(":memory:",
+		metric.WithMaxOpenConns(1),
+		metric.WithRollupPolicy(defaultRollupPolicy()),
+	))
+	if err != nil {
+		t.Fatalf("open metric store: %v", err)
+	}
+	defer s.Close()
+	if err := createMetricDefinitions(ctx, s); err != nil {
+		t.Fatalf("create metric definitions: %v", err)
+	}
+
+	base := time.Now().UTC().Truncate(time.Minute).Add(-time.Minute)
+	if err := s.WriteBatch(ctx, []metric.Point{
+		{MetricName: MetricCPU, EntityID: "node-a", Timestamp: base.Add(10 * time.Second), Value: 10},
+		{MetricName: MetricCPU, EntityID: "node-a", Timestamp: base.Add(20 * time.Second), Value: 90},
+		{MetricName: MetricRAM, EntityID: "node-a", Timestamp: base.Add(20 * time.Second), Value: 123456},
+	}); err != nil {
+		t.Fatalf("write metric points: %v", err)
+	}
+
+	got, err := getRecordMetricMaxByClientAndTimeFromSeries(ctx, s, "node-a", "cpu", base, base.Add(time.Minute))
+	if err != nil {
+		t.Fatalf("get CPU max records: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("record count = %d, want 1: %#v", len(got), got)
+	}
+	if got[0].Cpu != 90 {
+		t.Fatalf("CPU max = %v, want 90", got[0].Cpu)
+	}
+	if got[0].Ram != 0 {
+		t.Fatalf("unselected RAM value = %d, want 0", got[0].Ram)
+	}
+}

database/metricstore/report.go

@@ -254,7 +254,14 @@ func drainReportQueue(queue <-chan v1.Report, limit int) []v1.Report {
 	if limit <= 0 {
 		return nil
 	}
-	reports := make([]v1.Report, 0, limit)
+	capacity := len(queue)
+	if capacity > limit {
+		capacity = limit
+	}
+	if capacity == 0 {
+		return nil
+	}
+	reports := make([]v1.Report, 0, capacity)
 	for len(reports) < limit {
 		select {
 		case report := <-queue:
@@ -270,7 +277,14 @@ func drainPingQueue(queue <-chan models.PingRecord, limit int) []models.PingReco
 	if limit <= 0 {
 		return nil
 	}
-	records := make([]models.PingRecord, 0, limit)
+	capacity := len(queue)
+	if capacity > limit {
+		capacity = limit
+	}
+	if capacity == 0 {
+		return nil
+	}
+	records := make([]models.PingRecord, 0, capacity)
 	for len(records) < limit {
 		select {
 		case record := <-queue:

database/records/records.go

@@ -28,6 +28,11 @@ func GetRecordsByClientAndTime(uuid string, start, end time.Time) ([]models.Reco
 	return metricstore.GetRecordsByClientAndTime(context.Background(), uuid, start, end)
 }
 
+// GetRecordMetricMaxByClientAndTime 获取单项监控指标在各时间桶内的最大值。
+func GetRecordMetricMaxByClientAndTime(uuid, metricName string, start, end time.Time) ([]models.Record, error) {
+	return metricstore.GetRecordMetricMaxByClientAndTime(context.Background(), uuid, metricName, start, end)
+}
+
 // GetRecordsByTime 获取所有客户端在时间范围内的记录。
 func GetRecordsByTime(start, end time.Time) ([]models.Record, error) {
 	return metricstore.GetRecordsByTime(context.Background(), start, end)

pkg/metric/raw_points.go

@@ -588,11 +588,14 @@ func (s *Store) latestRollupBefore(ctx context.Context, metricName, entityID str
 		var timestamp int64
 		var rawTags, rawLabels any
 		err := s.reader().QueryRowContext(ctx, fmt.Sprintf(`SELECT r.last_val, r.last_ts_milli, s.tags, l.labels
-			FROM %s r JOIN %s s ON s.id = r.series_id JOIN %s d ON d.id = r.resolution_id JOIN %s l ON l.id = r.label_id
-			WHERE s.metric_name = %s AND s.entity_id = %s AND d.resolution_milli = %s AND r.last_ts_milli <= %s
+			FROM %s r JOIN %s s ON s.id = r.series_id JOIN %s l ON l.id = r.label_id
+			WHERE r.series_id IN (SELECT id FROM %s WHERE metric_name = %s AND entity_id = %s)
+			AND r.resolution_id IN (SELECT id FROM %s WHERE resolution_milli = %s)
+			AND r.last_ts_milli <= %s
 			ORDER BY r.last_ts_milli DESC LIMIT 1`,
-			s.tables.rollups, s.tables.series, s.tables.resolutions, s.tables.labels,
-			s.dialect.placeholder(1), s.dialect.placeholder(2), s.dialect.placeholder(3), s.dialect.placeholder(4)),
+			s.tables.rollups, s.tables.series, s.tables.labels, s.tables.series,
+			s.dialect.placeholder(1), s.dialect.placeholder(2), s.tables.resolutions,
+			s.dialect.placeholder(3), s.dialect.placeholder(4)),
 			metricName, entityID, tier.Interval.Milliseconds(), endMilli,
 		).Scan(&value, &timestamp, &rawTags, &rawLabels)
 		if err == sql.ErrNoRows {

pkg/metric/rollup_query.go

@@ -325,25 +325,27 @@ func (s *Store) coveringSeriesResolution(ctx context.Context, query Query, prefe
 }
 
 func (s *Store) rollupTimeBounds(ctx context.Context, query Query, interval time.Duration) (int64, int64, bool, error) {
-	args := []any{query.MetricName, interval.Milliseconds(), query.End.UTC().UnixMilli()}
-	parts := []string{
-		"s.metric_name = " + s.dialect.placeholder(1),
-		"d.resolution_milli = " + s.dialect.placeholder(2),
-		"r.first_ts_milli <= " + s.dialect.placeholder(3),
-	}
+	args := []any{query.MetricName}
+	seriesParts := []string{"s.metric_name = " + s.dialect.placeholder(1)}
 	if query.EntityID != "" {
 		args = append(args, query.EntityID)
-		parts = append(parts, "s.entity_id = "+s.dialect.placeholder(len(args)))
+		seriesParts = append(seriesParts, "s.entity_id = "+s.dialect.placeholder(len(args)))
 	}
 	for _, key := range sortedKeys(query.Tags) {
 		args = append(args, query.Tags[key])
-		parts = append(parts, s.dialect.jsonExtractEquals("s.tags", key, s.dialect.placeholder(len(args))))
+		seriesParts = append(seriesParts, s.dialect.jsonExtractEquals("s.tags", key, s.dialect.placeholder(len(args))))
 	}
+	args = append(args, interval.Milliseconds())
+	resolutionPlaceholder := s.dialect.placeholder(len(args))
+	args = append(args, query.End.UTC().UnixMilli())
+	endPlaceholder := s.dialect.placeholder(len(args))
+
 	var earliest, latest sql.NullInt64
 	err := s.reader().QueryRowContext(ctx, fmt.Sprintf(`SELECT MIN(r.first_ts_milli), MAX(r.last_ts_milli) FROM %s r
-		JOIN %s s ON s.id = r.series_id
-		JOIN %s d ON d.id = r.resolution_id
-		WHERE %s`, s.tables.rollups, s.tables.series, s.tables.resolutions, joinSQLWith(parts, " AND ")),
+		WHERE r.series_id IN (SELECT s.id FROM %s s WHERE %s)
+		AND r.resolution_id IN (SELECT d.id FROM %s d WHERE d.resolution_milli = %s)
+		AND r.first_ts_milli <= %s`, s.tables.rollups, s.tables.series, joinSQLWith(seriesParts, " AND "),
+		s.tables.resolutions, resolutionPlaceholder, endPlaceholder),
 		args...).Scan(&earliest, &latest)
 	if err != nil {
 		return 0, 0, false, err

utils/notifier/load.go

@@ -2,7 +2,6 @@ package notifier
 
 import (
 	"fmt"
-	logger "github.com/komari-monitor/komari/utils/log"
 	"reflect"
 	"sync"
 	"time"
@@ -13,6 +12,7 @@ import (
 	messageevent "github.com/komari-monitor/komari/database/models/messageEvent"
 	"github.com/komari-monitor/komari/database/records"
 	"github.com/komari-monitor/komari/internal/scheduler"
+	logger "github.com/komari-monitor/komari/utils/log"
 	"github.com/komari-monitor/komari/utils/messageSender"
 )
 
@@ -68,8 +68,8 @@ func executeLoadNotificationTask(task models.LoadNotification) {
 	windowStart := now.Add(-time.Duration(task.Interval) * time.Minute)
 	overloadClients := make([]string, 0)
 	for _, clientUUID := range task.Clients {
-		// 获取客户端在时间窗口内的记录
-		records, err := getRecordsForClient(clientUUID, windowStart, now)
+		// 仅查询当前通知使用的指标，避免重建完整监控记录。
+		records, err := getMetricRecordsForClient(clientUUID, task.Metric, windowStart, now)
 		if err != nil {
 			continue
 		}
@@ -97,9 +97,9 @@ func shouldSkipNotification(task models.LoadNotification) bool {
 	return timeSinceLastNotified < cooldownPeriod
 }
 
-// getRecordsForClient 获取指定客户端在时间窗口内的记录
-func getRecordsForClient(clientUUID string, start, end time.Time) ([]models.Record, error) {
-	return records.GetRecordsByClientAndTime(clientUUID, start, end)
+// getMetricRecordsForClient 获取指定客户端在时间窗口内的单项指标最大值。
+func getMetricRecordsForClient(clientUUID, metricName string, start, end time.Time) ([]models.Record, error) {
+	return records.GetRecordMetricMaxByClientAndTime(clientUUID, metricName, start, end)
 }
 
 // checkMetricThreshold 检查指标是否达到阈值
@@ -143,7 +143,7 @@ func getMetricValue(record models.Record, metric string) float32 {
 			logger.Errorf("notifier", "Failed to get client info for %s: %v", record.Client, err)
 			return 0
 		}
-		if record.RamTotal > 0 {
+		if client.MemTotal > 0 {
 			return float32(record.Ram) / float32(client.MemTotal) * 100
 		}
 		return 0
@@ -153,7 +153,7 @@ func getMetricValue(record models.Record, metric string) float32 {
 			logger.Errorf("notifier", "Failed to get client info for %s: %v", record.Client, err)
 			return 0
 		}
-		if record.SwapTotal > 0 {
+		if client.SwapTotal > 0 {
 			return float32(record.Swap) / float32(client.SwapTotal) * 100
 		}
 		return 0
@@ -167,7 +167,7 @@ func getMetricValue(record models.Record, metric string) float32 {
 			logger.Errorf("notifier", "Failed to get client info for %s: %v", record.Client, err)
 			return 0
 		}
-		if record.DiskTotal > 0 {
+		if client.DiskTotal > 0 {
 			return float32(record.Disk) / float32(client.DiskTotal) * 100
 		}
 		return 0