[功能] 实现节点访问日志的分片支持，优化日志查询和管理逻辑

openflare_server/model/node_access_log.go

@@ -2,6 +2,7 @@ package model
 
 import (
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -10,13 +11,13 @@ import (
 
 type NodeAccessLog struct {
 	ID         uint      `json:"id" gorm:"primaryKey"`
-	NodeID     string    `json:"node_id" gorm:"index:idx_node_access_logs_node_logged_at,priority:1;size:64;not null"`
-	LoggedAt   time.Time `json:"logged_at" gorm:"index:idx_node_access_logs_logged_at;index:idx_node_access_logs_node_logged_at,priority:2"`
-	RemoteAddr string    `json:"remote_addr" gorm:"index:idx_node_access_logs_remote_addr;size:128"`
+	NodeID     string    `json:"node_id" gorm:"index:,composite:node_logged_at,priority:1;size:64;not null"`
+	LoggedAt   time.Time `json:"logged_at" gorm:"index;index:,composite:node_logged_at,priority:2"`
+	RemoteAddr string    `json:"remote_addr" gorm:"index;size:128"`
 	Region     string    `json:"region" gorm:"size:128"`
-	Host       string    `json:"host" gorm:"index:idx_node_access_logs_host;size:255"`
+	Host       string    `json:"host" gorm:"index;size:255"`
 	Path       string    `json:"path" gorm:"size:2048"`
-	StatusCode int       `json:"status_code" gorm:"index:idx_node_access_logs_status_code"`
+	StatusCode int       `json:"status_code" gorm:"index"`
 	RawJSON    string    `json:"raw_json" gorm:"type:text"`
 	CreatedAt  time.Time `json:"created_at"`
 }
@@ -93,84 +94,87 @@ type NodeAccessLogTrendPointRow struct {
 }
 
 func ListNodeAccessLogs(query NodeAccessLogQuery) (logs []*NodeAccessLog, err error) {
-	offset := query.Page * query.PageSize
-	db := buildNodeAccessLogQuery(DB, query).
-		Order(buildNodeAccessLogSortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset)
-	err = db.Find(&logs).Error
-	return logs, err
+	all, err := listNodeAccessLogsAcrossShards(query)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(all), query.Page, query.PageSize)
+	if start >= len(all) {
+		return []*NodeAccessLog{}, nil
+	}
+	return all[start:end], nil
 }
 
 func CountNodeAccessLogs(query NodeAccessLogQuery) (totalRecords int64, totalIPs int64, err error) {
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), query)
-	if err = base.Count(&totalRecords).Error; err != nil {
+	all, err := listNodeAccessLogsAcrossShards(query)
+	if err != nil {
 		return 0, 0, err
 	}
-	distinctQuery := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), query).
-		Where("remote_addr <> ''").
-		Distinct("remote_addr")
-	if err = distinctQuery.Count(&totalIPs).Error; err != nil {
-		return 0, 0, err
+	ips := make(map[string]struct{}, len(all))
+	for _, item := range all {
+		if item == nil {
+			continue
+		}
+		trimmed := strings.TrimSpace(item.RemoteAddr)
+		if trimmed != "" {
+			ips[trimmed] = struct{}{}
+		}
 	}
-	return totalRecords, totalIPs, nil
+	return int64(len(all)), int64(len(ips)), nil
 }
 
 func ListNodeAccessLogRegionCounts(nodeID string, since time.Time, limit int) (items []*NodeAccessLogRegionCount, err error) {
-	query := DB.Model(&NodeAccessLog{}).
-		Select("region as region, count(*) as count").
-		Where("region <> ''")
-	if nodeID != "" {
-		query = query.Where("node_id = ?", nodeID)
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID: nodeID,
+		Since:  since,
+	})
+	if err != nil {
+		return nil, err
 	}
-	if !since.IsZero() {
-		query = query.Where("logged_at >= ?", since)
+	counts := make(map[string]int64)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		region := strings.TrimSpace(item.Region)
+		if region == "" {
+			continue
+		}
+		counts[region]++
 	}
-	query = query.Group("region").Order("count desc, region asc")
-	if limit > 0 {
-		query = query.Limit(limit)
+	items = make([]*NodeAccessLogRegionCount, 0, len(counts))
+	for region, count := range counts {
+		items = append(items, &NodeAccessLogRegionCount{
+			Region: region,
+			Count:  count,
+		})
 	}
-	err = query.Scan(&items).Error
-	return items, err
+	sort.Slice(items, func(i int, j int) bool {
+		if items[i].Count == items[j].Count {
+			return items[i].Region < items[j].Region
+		}
+		return items[i].Count > items[j].Count
+	})
+	if limit > 0 && len(items) > limit {
+		items = items[:limit]
+	}
+	return items, nil
 }
 
 func ListNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (items []*NodeAccessLogBucketRow, err error) {
-	offset := query.Page * query.PageSize
-	bucketExpr := accessLogBucketEpochExpr(query.FoldMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Path:       query.Path,
-		Since:      query.Since,
-	})
-	err = base.Select(fmt.Sprintf(
-		"%s as bucket_epoch, count(*) as request_count, count(distinct remote_addr) as unique_ip_count, count(distinct host) as unique_host_count, sum(case when status_code < 400 then 1 else 0 end) as success_count, sum(case when status_code >= 400 and status_code < 500 then 1 else 0 end) as client_error_count, sum(case when status_code >= 500 then 1 else 0 end) as server_error_count",
-		bucketExpr,
-	)).
-		Group(bucketExpr).
-		Order(buildNodeAccessLogBucketSortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset).
-		Scan(&items).Error
-	return items, err
+	rows, err := buildNodeAccessLogBucketRows(query)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(rows), query.Page, query.PageSize)
+	if start >= len(rows) {
+		return []*NodeAccessLogBucketRow{}, nil
+	}
+	return rows[start:end], nil
 }
 
 func CountNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (total int64, err error) {
-	bucketExpr := accessLogBucketEpochExpr(query.FoldMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Path:       query.Path,
-		Since:      query.Since,
-	})
-	rows := []struct {
-		BucketEpoch int64 `gorm:"column:bucket_epoch"`
-	}{}
-	err = base.Select(fmt.Sprintf("%s as bucket_epoch", bucketExpr)).
-		Group(bucketExpr).
-		Scan(&rows).Error
+	rows, err := buildNodeAccessLogBucketRows(query)
 	if err != nil {
 		return 0, err
 	}
@@ -178,39 +182,19 @@ func CountNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (total int64, err
 }
 
 func ListNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery, recentSince time.Time) (items []*NodeAccessLogIPSummaryRow, err error) {
-	offset := query.Page * query.PageSize
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Since:      query.Since,
-	}).Where("remote_addr <> ''")
-	lastSeenExpr := accessLogEpochExpr("max(logged_at)")
-	err = base.Select(
-		"remote_addr as remote_addr, count(*) as total_requests, sum(case when logged_at >= ? then 1 else 0 end) as recent_requests, "+lastSeenExpr+" as last_seen_epoch",
-		recentSince,
-	).
-		Group("remote_addr").
-		Order(buildNodeAccessLogIPSummarySortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset).
-		Scan(&items).Error
-	return items, err
+	rows, err := buildNodeAccessLogIPSummaryRows(query, recentSince)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(rows), query.Page, query.PageSize)
+	if start >= len(rows) {
+		return []*NodeAccessLogIPSummaryRow{}, nil
+	}
+	return rows[start:end], nil
 }
 
 func CountNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery) (total int64, err error) {
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Since:      query.Since,
-	}).Where("remote_addr <> ''")
-	rows := []struct {
-		RemoteAddr string `gorm:"column:remote_addr"`
-	}{}
-	err = base.Select("remote_addr").
-		Group("remote_addr").
-		Scan(&rows).Error
+	rows, err := buildNodeAccessLogIPSummaryRows(query, time.Time{})
 	if err != nil {
 		return 0, err
 	}
@@ -218,23 +202,49 @@ func CountNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery) (total int
 }
 
 func ListNodeAccessLogIPTrend(query NodeAccessLogIPTrendQuery) (items []*NodeAccessLogTrendPointRow, err error) {
-	bucketExpr := accessLogBucketEpochExpr(query.BucketMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
 		NodeID:     query.NodeID,
 		RemoteAddr: query.RemoteAddr,
 		Host:       query.Host,
 		Since:      query.Since,
-	}).Where("remote_addr = ?", strings.TrimSpace(query.RemoteAddr))
-	err = base.Select(fmt.Sprintf("%s as bucket_epoch, count(*) as request_count", bucketExpr)).
-		Group(bucketExpr).
-		Order("bucket_epoch asc").
-		Scan(&items).Error
-	return items, err
+	})
+	if err != nil {
+		return nil, err
+	}
+	remoteAddr := strings.TrimSpace(query.RemoteAddr)
+	if remoteAddr == "" {
+		return []*NodeAccessLogTrendPointRow{}, nil
+	}
+	buckets := make(map[int64]int64)
+	for _, item := range logs {
+		if item == nil || strings.TrimSpace(item.RemoteAddr) != remoteAddr {
+			continue
+		}
+		bucketEpoch := bucketEpochForTime(item.LoggedAt, query.BucketMinutes)
+		buckets[bucketEpoch]++
+	}
+	items = make([]*NodeAccessLogTrendPointRow, 0, len(buckets))
+	for bucketEpoch, requestCount := range buckets {
+		items = append(items, &NodeAccessLogTrendPointRow{
+			BucketEpoch:  bucketEpoch,
+			RequestCount: requestCount,
+		})
+	}
+	sort.Slice(items, func(i int, j int) bool {
+		return items[i].BucketEpoch < items[j].BucketEpoch
+	})
+	return items, nil
 }
 
 func DeleteNodeAccessLogsBefore(before time.Time) (deleted int64, err error) {
-	result := DB.Where("logged_at < ?", before).Delete(&NodeAccessLog{})
-	return result.RowsAffected, result.Error
+	for _, table := range observabilityShardTables("node_access_logs") {
+		result := DB.Table(table).Where("logged_at < ?", before).Delete(&NodeAccessLog{})
+		if result.Error != nil {
+			return deleted, result.Error
+		}
+		deleted += result.RowsAffected
+	}
+	return deleted, nil
 }
 
 func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
@@ -244,6 +254,10 @@ func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	if db.Statement == nil || db.Statement.Model == nil {
 		db = db.Model(&NodeAccessLog{})
 	}
+	return applyNodeAccessLogFilters(db, query)
+}
+
+func applyNodeAccessLogFilters(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	if trimmed := strings.TrimSpace(query.NodeID); trimmed != "" {
 		db = db.Where("node_id LIKE ?", "%"+trimmed+"%")
 	}
@@ -262,6 +276,297 @@ func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	return db
 }
 
+func listNodeAccessLogsAcrossShards(query NodeAccessLogQuery) ([]*NodeAccessLog, error) {
+	items, err := queryAcrossShards("node_access_logs", func(tx *gorm.DB) ([]*NodeAccessLog, error) {
+		var shardRows []*NodeAccessLog
+		if err := applyNodeAccessLogFilters(tx, query).Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	sortNodeAccessLogs(items, query.SortBy, query.SortOrder)
+	return items, nil
+}
+
+func buildNodeAccessLogBucketRows(query NodeAccessLogBucketQuery) ([]*NodeAccessLogBucketRow, error) {
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID:     query.NodeID,
+		RemoteAddr: query.RemoteAddr,
+		Host:       query.Host,
+		Path:       query.Path,
+		Since:      query.Since,
+	})
+	if err != nil {
+		return nil, err
+	}
+	type bucketAccumulator struct {
+		requestCount     int64
+		uniqueIPs        map[string]struct{}
+		uniqueHosts      map[string]struct{}
+		successCount     int64
+		clientErrorCount int64
+		serverErrorCount int64
+	}
+	accumulators := make(map[int64]*bucketAccumulator)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		bucketEpoch := bucketEpochForTime(item.LoggedAt, query.FoldMinutes)
+		accumulator := accumulators[bucketEpoch]
+		if accumulator == nil {
+			accumulator = &bucketAccumulator{
+				uniqueIPs:   make(map[string]struct{}),
+				uniqueHosts: make(map[string]struct{}),
+			}
+			accumulators[bucketEpoch] = accumulator
+		}
+		accumulator.requestCount++
+		if trimmed := strings.TrimSpace(item.RemoteAddr); trimmed != "" {
+			accumulator.uniqueIPs[trimmed] = struct{}{}
+		}
+		if trimmed := strings.TrimSpace(item.Host); trimmed != "" {
+			accumulator.uniqueHosts[trimmed] = struct{}{}
+		}
+		switch {
+		case item.StatusCode < 400:
+			accumulator.successCount++
+		case item.StatusCode < 500:
+			accumulator.clientErrorCount++
+		default:
+			accumulator.serverErrorCount++
+		}
+	}
+	rows := make([]*NodeAccessLogBucketRow, 0, len(accumulators))
+	for bucketEpoch, accumulator := range accumulators {
+		rows = append(rows, &NodeAccessLogBucketRow{
+			BucketEpoch:      bucketEpoch,
+			RequestCount:     accumulator.requestCount,
+			UniqueIPCount:    int64(len(accumulator.uniqueIPs)),
+			UniqueHostCount:  int64(len(accumulator.uniqueHosts)),
+			SuccessCount:     accumulator.successCount,
+			ClientErrorCount: accumulator.clientErrorCount,
+			ServerErrorCount: accumulator.serverErrorCount,
+		})
+	}
+	sortNodeAccessLogBucketRows(rows, query.SortBy, query.SortOrder)
+	return rows, nil
+}
+
+func buildNodeAccessLogIPSummaryRows(query NodeAccessLogIPSummaryQuery, recentSince time.Time) ([]*NodeAccessLogIPSummaryRow, error) {
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID:     query.NodeID,
+		RemoteAddr: query.RemoteAddr,
+		Host:       query.Host,
+		Since:      query.Since,
+	})
+	if err != nil {
+		return nil, err
+	}
+	type accumulator struct {
+		totalRequests  int64
+		recentRequests int64
+		lastSeenAt     time.Time
+	}
+	accumulators := make(map[string]*accumulator)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		remoteAddr := strings.TrimSpace(item.RemoteAddr)
+		if remoteAddr == "" {
+			continue
+		}
+		acc := accumulators[remoteAddr]
+		if acc == nil {
+			acc = &accumulator{}
+			accumulators[remoteAddr] = acc
+		}
+		acc.totalRequests++
+		if !recentSince.IsZero() && !item.LoggedAt.Before(recentSince) {
+			acc.recentRequests++
+		}
+		if item.LoggedAt.After(acc.lastSeenAt) {
+			acc.lastSeenAt = item.LoggedAt
+		}
+	}
+	rows := make([]*NodeAccessLogIPSummaryRow, 0, len(accumulators))
+	for remoteAddr, acc := range accumulators {
+		rows = append(rows, &NodeAccessLogIPSummaryRow{
+			RemoteAddr:     remoteAddr,
+			TotalRequests:  acc.totalRequests,
+			RecentRequests: acc.recentRequests,
+			LastSeenEpoch:  acc.lastSeenAt.Unix(),
+		})
+	}
+	sortNodeAccessLogIPSummaryRows(rows, query.SortBy, query.SortOrder)
+	return rows, nil
+}
+
+func sortNodeAccessLogs(items []*NodeAccessLog, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "status_code":
+			compare = compareInt(left.StatusCode, right.StatusCode)
+		case "remote_addr":
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		case "host":
+			compare = strings.Compare(left.Host, right.Host)
+		case "path":
+			compare = strings.Compare(left.Path, right.Path)
+		default:
+			compare = compareTime(left.LoggedAt, right.LoggedAt)
+		}
+		if compare == 0 {
+			compare = compareTime(left.LoggedAt, right.LoggedAt)
+		}
+		if compare == 0 {
+			compare = compareUint(left.ID, right.ID)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func sortNodeAccessLogBucketRows(items []*NodeAccessLogBucketRow, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "request_count":
+			compare = compareInt64(left.RequestCount, right.RequestCount)
+		default:
+			compare = compareInt64(left.BucketEpoch, right.BucketEpoch)
+		}
+		if compare == 0 {
+			compare = compareInt64(left.BucketEpoch, right.BucketEpoch)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func sortNodeAccessLogIPSummaryRows(items []*NodeAccessLogIPSummaryRow, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "recent_requests":
+			compare = compareInt64(left.RecentRequests, right.RecentRequests)
+		case "last_seen_at":
+			compare = compareInt64(left.LastSeenEpoch, right.LastSeenEpoch)
+		case "remote_addr":
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		default:
+			compare = compareInt64(left.TotalRequests, right.TotalRequests)
+		}
+		if compare == 0 {
+			compare = compareInt64(left.LastSeenEpoch, right.LastSeenEpoch)
+		}
+		if compare == 0 {
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func paginateBounds(total int, page int, pageSize int) (int, int) {
+	if page < 0 {
+		page = 0
+	}
+	if pageSize <= 0 {
+		return 0, total
+	}
+	start := page * pageSize
+	if start > total {
+		start = total
+	}
+	end := start + pageSize
+	if end > total {
+		end = total
+	}
+	return start, end
+}
+
+func bucketEpochForTime(value time.Time, bucketMinutes int) int64 {
+	bucketSeconds := int64(bucketMinutes * 60)
+	if bucketSeconds <= 0 {
+		bucketSeconds = 180
+	}
+	return (value.UTC().Unix() / bucketSeconds) * bucketSeconds
+}
+
+func compareTime(left time.Time, right time.Time) int {
+	switch {
+	case left.After(right):
+		return 1
+	case left.Before(right):
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareInt(left int, right int) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareInt64(left int64, right int64) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareUint(left uint, right uint) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
 func buildNodeAccessLogSortClause(sortBy string, sortOrder string) string {
 	column := "logged_at"
 	switch strings.TrimSpace(sortBy) {