tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-10-04 15:06:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(dashboard): 首页请求趋势拆分状态码并合并容量到业务流量

Browse Source 
- 24 小时请求趋势拆分展示请求总量与 200/400/500 状态码请求量，独占一行；
  时间桶聚合新增 status_200/400/500_count（CH countIf、PG FILTER），
  请求趋势改为基于原始桶聚合（小时 rollup 无状态码口径）
- 首页移除宿主机磁盘指标，容量趋势（CPU/内存）并入业务流量卡片展示
- 压缩协议 traffic_24h 扩展为 7 元组，前端归一化同步更新
This commit is contained in:
ryan
2026-08-13 10:57:21 +08:00
parent a9e4237bbf
commit a4dd5ca9e1
15 changed files with 196 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/apps/openflare/dashboard/logics.go
+4
View File
@@ -330,11 +330,15 @@ func compressDistributionItems(items []observability.DistributionItem) [][]any {
func compressTrafficTrendPoints(points []observability.TrafficTrendPoint) [][]any {
rows := make([][]any, 0, len(points))
for _, point := range points {
// Compact layout: [0] bucket, [1] request, [2] error, [3] uv, [4] 2xx, [5] 4xx, [6] 5xx
rows = append(rows, []any{
point.BucketStartedAt,
point.RequestCount,
point.ErrorCount,
point.UniqueVisitorCount,
point.Status2xxCount,
point.Status4xxCount,
point.Status5xxCount,
})
}
return rows
internal/apps/openflare/dashboard/logics_test.go
+1 -1
View File
@@ -143,7 +143,7 @@ func TestGetOverviewStructure(t *testing.T) {
require.Len(t, overview.Trends.Network24h, 24)
require.Len(t, overview.Trends.DiskIO24h, 24)
for _, row := range overview.Trends.Traffic24h {
require.Len(t, row, 4)
require.Len(t, row, 7)
}
for _, row := range overview.Trends.Capacity24h {
require.Len(t, row, 4)
internal/apps/openflare/observability/analytics.go
+10 -19
View File
@@ -85,6 +85,9 @@ type TrafficTrendPoint struct {
RequestCount int64 `json:"request_count"`
ErrorCount int64 `json:"error_count"`
UniqueVisitorCount int64 `json:"unique_visitor_count"`
Status2xxCount int64 `json:"status_2xx_count"`
Status4xxCount int64 `json:"status_4xx_count"`
Status5xxCount int64 `json:"status_5xx_count"`
}
// CapacityTrendPoint is a capacity trend bucket.
@@ -303,9 +306,10 @@ func BuildNodeTrends(
}
}
// BuildTrafficTrendPointsFromAccessLogs builds 24h request/error buckets from access logs.
// Prefers of_access_log_hourly when available; falls back to raw bucket aggregates.
// UniqueVisitorCount on hourly path is 0 (use TrafficSummary for exact UV).
// BuildTrafficTrendPointsFromAccessLogs builds 24h request/error/status buckets from access logs.
// Uses raw bucket aggregates: the hourly rollup (of_access_log_hourly) has no per-status counts,
// and the 24h window on the dashboard is cached, so the raw scan is acceptable.
// UniqueVisitorCount from buckets is exact (uniqExact on raw); TrafficSummary is used elsewhere for UV.
func BuildTrafficTrendPointsFromAccessLogs(ctx context.Context, now time.Time, nodeID string, since time.Time) []TrafficTrendPoint {
start := trendWindowStart(now)
points := make([]TrafficTrendPoint, observabilityTrendBuckets)
@@ -313,22 +317,6 @@ func BuildTrafficTrendPointsFromAccessLogs(ctx context.Context, now time.Time, n
points[index].BucketStartedAt = start.Add(time.Duration(index) * time.Hour)
}
if hourly, err := repository.ListOpenFlareTrafficHourlySince(ctx, nodeID, since); err == nil && len(hourly) > 0 {
for _, row := range hourly {
if row == nil {
continue
}
index, ok := trendBucketIndex(row.Hour, start)
if !ok {
continue
}
points[index].RequestCount += row.RequestCount
points[index].ErrorCount += row.ErrorCount
// UniqueVisitorCount intentionally not summed from hourly rollup (always 0 / overcounts).
}
return points
}
buckets, err := repository.ListOpenFlareAccessLogBuckets(ctx, model.OpenFlareAccessLogBucketQuery{
NodeID: nodeID,
Since: since,
@@ -353,6 +341,9 @@ func BuildTrafficTrendPointsFromAccessLogs(ctx context.Context, now time.Time, n
points[index].RequestCount = row.RequestCount
points[index].ErrorCount = row.ServerErrorCount
points[index].UniqueVisitorCount = row.UniqueIPCount
points[index].Status2xxCount = row.Status2xxCount
points[index].Status4xxCount = row.Status4xxCount
points[index].Status5xxCount = row.Status5xxCount
}
}
return points
internal/model/analytics/node_access_log_stats.go
+3
View File
@@ -10,6 +10,9 @@ type NodeAccessLogBucketAggregate struct {
SuccessCount int64 `gorm:"column:success_count"`
ClientErrorCount int64 `gorm:"column:client_error_count"`
ServerErrorCount int64 `gorm:"column:server_error_count"`
Status2xxCount int64 `gorm:"column:status_2xx_count"`
Status4xxCount int64 `gorm:"column:status_4xx_count"`
Status5xxCount int64 `gorm:"column:status_5xx_count"`
UniqueIPCount int64 `gorm:"column:unique_ip_count"`
UniqueHostCount int64 `gorm:"column:unique_host_count"`
BytesSent int64 `gorm:"column:bytes_sent"`
internal/model/openflare_observability.go
+3
View File
@@ -198,6 +198,9 @@ type OpenFlareAccessLogBucketRow struct {
SuccessCount int64 `json:"success_count"`
ClientErrorCount int64 `json:"client_error_count"`
ServerErrorCount int64 `json:"server_error_count"`
Status2xxCount int64 `json:"status_2xx_count"`
Status4xxCount int64 `json:"status_4xx_count"`
Status5xxCount int64 `json:"status_5xx_count"`
BytesSent int64 `json:"bytes_sent"`
RequestLength int64 `json:"request_length"`
}
internal/repository/analytics/node_access_log_stats.go
+9 -3
View File
@@ -46,6 +46,9 @@ SELECT
countIf(status_code < 400) AS success_count,
countIf(status_code >= 400 AND status_code < 500) AS client_error_count,
countIf(status_code >= 500) AS server_error_count,
countIf(status_code >= 200 AND status_code < 300) AS status_2xx_count,
countIf(status_code >= 400 AND status_code < 500) AS status_4xx_count,
countIf(status_code >= 500) AS status_5xx_count,
uniqExactIf(remote_addr, remote_addr != '') AS unique_ip_count,
uniqExactIf(host, host != '') AS unique_host_count,
sum(bytes_sent) AS bytes_sent,
@@ -70,10 +73,10 @@ ORDER BY %s`, bucketExpr, tableName, clause, nodeAccessLogBucketOrderClause(filt
var result []NodeAccessLogBucketAggregate
for rows.Next() {
var (
bucketEpoch int64
requestCount, successCount, clientErrorCount, serverErrorCount, uniqueIPCount, uniqueHostCount, bytesSent, requestLength uint64
bucketEpoch int64
requestCount, successCount, clientErrorCount, serverErrorCount, status2xxCount, status4xxCount, status5xxCount, uniqueIPCount, uniqueHostCount, bytesSent, requestLength uint64
)
if err := rows.Scan(&bucketEpoch, &requestCount, &successCount, &clientErrorCount, &serverErrorCount, &uniqueIPCount, &uniqueHostCount, &bytesSent, &requestLength); err != nil {
if err := rows.Scan(&bucketEpoch, &requestCount, &successCount, &clientErrorCount, &serverErrorCount, &status2xxCount, &status4xxCount, &status5xxCount, &uniqueIPCount, &uniqueHostCount, &bytesSent, &requestLength); err != nil {
return nil, fmt.Errorf("scan bucket aggregate row: %w", err)
}
result = append(result, NodeAccessLogBucketAggregate{
@@ -82,6 +85,9 @@ ORDER BY %s`, bucketExpr, tableName, clause, nodeAccessLogBucketOrderClause(filt
SuccessCount: safeInt64Count(successCount),
ClientErrorCount: safeInt64Count(clientErrorCount),
ServerErrorCount: safeInt64Count(serverErrorCount),
Status2xxCount: safeInt64Count(status2xxCount),
Status4xxCount: safeInt64Count(status4xxCount),
Status5xxCount: safeInt64Count(status5xxCount),
UniqueIPCount: safeInt64Count(uniqueIPCount),
UniqueHostCount: safeInt64Count(uniqueHostCount),
BytesSent: safeInt64Count(bytesSent),
internal/repository/logstore/postgres_store.go
+9
View File
@@ -307,6 +307,9 @@ func (s *gormLogStore) BucketAggregates(ctx context.Context, query model.OpenFla
SuccessCount int64
ClientErrorCount int64
ServerErrorCount int64
Status2xxCount int64 `gorm:"column:status_2xx_count"`
Status4xxCount int64 `gorm:"column:status_4xx_count"`
Status5xxCount int64 `gorm:"column:status_5xx_count"`
UniqueIPCount int64
UniqueHostCount int64
BytesSent int64
@@ -322,6 +325,9 @@ func (s *gormLogStore) BucketAggregates(ctx context.Context, query model.OpenFla
"COUNT(*) FILTER (WHERE status_code < 400) AS success_count, " +
"COUNT(*) FILTER (WHERE status_code >= 400 AND status_code < 500) AS client_error_count, " +
"COUNT(*) FILTER (WHERE status_code >= 500) AS server_error_count, " +
"COUNT(*) FILTER (WHERE status_code >= 200 AND status_code < 300) AS status_2xx_count, " +
"COUNT(*) FILTER (WHERE status_code >= 400 AND status_code < 500) AS status_4xx_count, " +
"COUNT(*) FILTER (WHERE status_code >= 500) AS status_5xx_count, " +
distinctNonEmptyCountSQL(s.db, "remote_addr") + " AS unique_ip_count, " +
distinctNonEmptyCountSQL(s.db, "host") + " AS unique_host_count, " +
"COALESCE(SUM(bytes_sent),0) AS bytes_sent, " +
@@ -338,6 +344,9 @@ func (s *gormLogStore) BucketAggregates(ctx context.Context, query model.OpenFla
SuccessCount: r.SuccessCount,
ClientErrorCount: r.ClientErrorCount,
ServerErrorCount: r.ServerErrorCount,
Status2xxCount: r.Status2xxCount,
Status4xxCount: r.Status4xxCount,
Status5xxCount: r.Status5xxCount,
UniqueIPCount: r.UniqueIPCount,
UniqueHostCount: r.UniqueHostCount,
BytesSent: r.BytesSent,
internal/repository/logstore/postgres_store_test.go
+6 -2
View File
@@ -709,9 +709,9 @@ func TestGormBucketAggregatesFullFieldSet(t *testing.T) {
rows := []analyticsmodel.NodeAccessLog{
{ID: 1, NodeID: "n1", LoggedAt: base, RemoteAddr: "1.1.1.1", Host: "a.example.com", StatusCode: 200, BytesSent: 100, RequestLength: 10},
{ID: 2, NodeID: "n1", LoggedAt: base.Add(time.Minute), RemoteAddr: "1.1.1.1", Host: "a.example.com", StatusCode: 301, BytesSent: 200, RequestLength: 20},
{ID: 3, NodeID: "n1", LoggedAt: base.Add(2 * time.Minute), RemoteAddr: "2.2.2.2", Host: "b.example.com", StatusCode: 404, BytesSent: 300, RequestLength: 30},
{ID: 3, NodeID: "n1", LoggedAt: base.Add(2 * time.Minute), RemoteAddr: "2.2.2.2", Host: "b.example.com", StatusCode: 400, BytesSent: 300, RequestLength: 30},
{ID: 4, NodeID: "n1", LoggedAt: base.Add(3 * time.Minute), RemoteAddr: "", Host: "b.example.com", StatusCode: 500, BytesSent: 400, RequestLength: 40},
{ID: 5, NodeID: "n1", LoggedAt: base.Add(4 * time.Minute), RemoteAddr: "3.3.3.3", Host: "c.example.com", StatusCode: 502, BytesSent: 500, RequestLength: 50},
{ID: 5, NodeID: "n1", LoggedAt: base.Add(4 * time.Minute), RemoteAddr: "3.3.3.3", Host: "c.example.com", StatusCode: 500, BytesSent: 500, RequestLength: 50},
{ID: 6, NodeID: "n2", LoggedAt: base.Add(time.Hour), RemoteAddr: "9.9.9.9", Host: "d.example.com", StatusCode: 200, BytesSent: 999, RequestLength: 99},
}
if err := s.BatchInsertNodeAccessLogs(ctx, rows); err != nil {
@@ -741,6 +741,10 @@ func TestGormBucketAggregatesFullFieldSet(t *testing.T) {
if b.ServerErrorCount != 2 {
t.Errorf("server_error_count = %d, want 2", b.ServerErrorCount)
}
if b.Status2xxCount != 1 || b.Status4xxCount != 1 || b.Status5xxCount != 2 {
t.Errorf("status class counts = 2xx:%d 4xx:%d 5xx:%d, want 1/1/2 (200/301/400/500/500)",
b.Status2xxCount, b.Status4xxCount, b.Status5xxCount)
}
if b.UniqueIPCount != 3 {
t.Errorf("unique_ip_count = %d, want 3 (empty remote_addr excluded)", b.UniqueIPCount)
}
internal/repository/openflare_access_log.go
+3
View File
@@ -258,6 +258,9 @@ func buildOpenFlareAccessLogBucketRows(ctx context.Context, query model.OpenFlar
SuccessCount: partial.SuccessCount,
ClientErrorCount: partial.ClientErrorCount,
ServerErrorCount: partial.ServerErrorCount,
Status2xxCount: partial.Status2xxCount,
Status4xxCount: partial.Status4xxCount,
Status5xxCount: partial.Status5xxCount,
BytesSent: partial.BytesSent,
RequestLength: partial.RequestLength,
})