feat: 添加节点和仪表板的流量分析、趋势和健康状态功能

atsf_server/service/observability_trends.go

@@ -21,6 +21,22 @@ type CapacityTrendPoint struct {
 	ReportedNodes             int       `json:"reported_nodes"`
 }
 
+type NetworkTrendPoint struct {
+	BucketStartedAt  time.Time `json:"bucket_started_at"`
+	NetworkRxBytes   int64     `json:"network_rx_bytes"`
+	NetworkTxBytes   int64     `json:"network_tx_bytes"`
+	OpenrestyRxBytes int64     `json:"openresty_rx_bytes"`
+	OpenrestyTxBytes int64     `json:"openresty_tx_bytes"`
+	ReportedNodes    int       `json:"reported_nodes"`
+}
+
+type DiskIOTrendPoint struct {
+	BucketStartedAt time.Time `json:"bucket_started_at"`
+	DiskReadBytes   int64     `json:"disk_read_bytes"`
+	DiskWriteBytes  int64     `json:"disk_write_bytes"`
+	ReportedNodes   int       `json:"reported_nodes"`
+}
+
 type capacityTrendAccumulator struct {
 	cpuSum   float64
 	cpuCount int
@@ -29,6 +45,10 @@ type capacityTrendAccumulator struct {
 	nodes    map[string]struct{}
 }
 
+type snapshotTrendAccumulator struct {
+	nodes map[string]struct{}
+}
+
 func buildTrafficTrendPoints(now time.Time, reports []*model.NodeRequestReport) []TrafficTrendPoint {
 	start := trendWindowStart(now)
 	points := make([]TrafficTrendPoint, observabilityTrendBuckets)
@@ -89,6 +109,64 @@ func buildCapacityTrendPoints(now time.Time, snapshots []*model.NodeMetricSnapsh
 	return points
 }
 
+func buildNetworkTrendPoints(now time.Time, snapshots []*model.NodeMetricSnapshot) []NetworkTrendPoint {
+	start := trendWindowStart(now)
+	points := make([]NetworkTrendPoint, observabilityTrendBuckets)
+	accumulators := make([]snapshotTrendAccumulator, observabilityTrendBuckets)
+	for index := range points {
+		points[index].BucketStartedAt = start.Add(time.Duration(index) * time.Hour)
+		accumulators[index].nodes = make(map[string]struct{})
+	}
+
+	for _, snapshot := range snapshots {
+		index, ok := trendBucketIndex(snapshot.CapturedAt, start)
+		if !ok {
+			continue
+		}
+		points[index].NetworkRxBytes += snapshot.NetworkRxBytes
+		points[index].NetworkTxBytes += snapshot.NetworkTxBytes
+		points[index].OpenrestyRxBytes += snapshot.OpenrestyRxBytes
+		points[index].OpenrestyTxBytes += snapshot.OpenrestyTxBytes
+		if snapshot.NodeID != "" {
+			accumulators[index].nodes[snapshot.NodeID] = struct{}{}
+		}
+	}
+
+	for index := range points {
+		points[index].ReportedNodes = len(accumulators[index].nodes)
+	}
+
+	return points
+}
+
+func buildDiskIOTrendPoints(now time.Time, snapshots []*model.NodeMetricSnapshot) []DiskIOTrendPoint {
+	start := trendWindowStart(now)
+	points := make([]DiskIOTrendPoint, observabilityTrendBuckets)
+	accumulators := make([]snapshotTrendAccumulator, observabilityTrendBuckets)
+	for index := range points {
+		points[index].BucketStartedAt = start.Add(time.Duration(index) * time.Hour)
+		accumulators[index].nodes = make(map[string]struct{})
+	}
+
+	for _, snapshot := range snapshots {
+		index, ok := trendBucketIndex(snapshot.CapturedAt, start)
+		if !ok {
+			continue
+		}
+		points[index].DiskReadBytes += snapshot.DiskReadBytes
+		points[index].DiskWriteBytes += snapshot.DiskWriteBytes
+		if snapshot.NodeID != "" {
+			accumulators[index].nodes[snapshot.NodeID] = struct{}{}
+		}
+	}
+
+	for index := range points {
+		points[index].ReportedNodes = len(accumulators[index].nodes)
+	}
+
+	return points
+}
+
 func trendWindowStart(now time.Time) time.Time {
 	return now.Truncate(time.Hour).Add(-(observabilityTrendBuckets - 1) * time.Hour)
 }