From dd49b2777d68aa54144fbcc28ce5598f93e1c4b7 Mon Sep 17 00:00:00 2001 From: ryan Date: Thu, 19 Mar 2026 16:19:46 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=8A=9F=E8=83=BD]=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E8=AE=BF=E9=97=AE=E6=97=A5=E5=BF=97=E7=9A=84?= =?UTF-8?q?=E5=88=86=E7=89=87=E6=94=AF=E6=8C=81=EF=BC=8C=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=9F=A5=E8=AF=A2=E5=92=8C=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/observability/traffic.go | 16 +- .../internal/observability/traffic_test.go | 26 + openflare_server/go.mod | 4 + openflare_server/go.sum | 10 + openflare_server/model/main.go | 24 +- openflare_server/model/main_test.go | 33 ++ openflare_server/model/node_access_log.go | 515 ++++++++++++++---- .../model/node_metric_snapshot.go | 32 +- openflare_server/model/node_request_report.go | 32 +- openflare_server/model/sharding.go | 81 +++ openflare_server/service/access_log_test.go | 63 ++- openflare_server/service/node_update_test.go | 16 +- openflare_server/service/observability.go | 9 +- 13 files changed, 709 insertions(+), 152 deletions(-) create mode 100644 openflare_server/model/sharding.go diff --git a/openflare_agent/internal/observability/traffic.go b/openflare_agent/internal/observability/traffic.go index 7a5edfc8..50a3a4da 100644 --- a/openflare_agent/internal/observability/traffic.go +++ b/openflare_agent/internal/observability/traffic.go @@ -327,18 +327,28 @@ type trafficCountItem struct { value int64 } +const accessLogPathMaxRunes = 100 + func normalizeAccessLogPath(value string) string { trimmed := strings.TrimSpace(value) if trimmed == "" { return "" } if strings.HasPrefix(trimmed, "http://") || strings.HasPrefix(trimmed, "https://") { - return trimmed + return truncateAccessLogPath(trimmed) } if strings.HasPrefix(trimmed, "/") { - return trimmed + return truncateAccessLogPath(trimmed) } - return "/" + trimmed + return truncateAccessLogPath("/" + trimmed) +} + +func truncateAccessLogPath(value string) string { + runes := []rune(value) + if len(runes) <= accessLogPathMaxRunes { + return value + } + return string(runes[:accessLogPathMaxRunes]) } func topCounts(values map[string]int64, limit int) map[string]int64 { diff --git a/openflare_agent/internal/observability/traffic_test.go b/openflare_agent/internal/observability/traffic_test.go index a5614606..c448f7de 100644 --- a/openflare_agent/internal/observability/traffic_test.go +++ b/openflare_agent/internal/observability/traffic_test.go @@ -3,6 +3,7 @@ package observability import ( "os" "path/filepath" + "strings" "testing" "openflare-agent/internal/config" @@ -108,6 +109,31 @@ func TestBuildTrafficObservabilityReturnsAccessLogs(t *testing.T) { } } +func TestBuildTrafficObservabilityTruncatesLongAccessLogPath(t *testing.T) { + tempDir := t.TempDir() + routeConfigPath := filepath.Join(tempDir, "conf.d", "openflare_routes.conf") + if err := os.MkdirAll(filepath.Dir(routeConfigPath), 0o755); err != nil { + t.Fatalf("MkdirAll failed: %v", err) + } + logPath := filepath.Join(filepath.Dir(routeConfigPath), "openflare_access.log") + longPath := "/" + strings.Repeat("a", 140) + content := []byte( + "{\"ts\":\"2026-03-14T08:00:00Z\",\"host\":\"app.example.com\",\"path\":\"" + longPath + "\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n", + ) + if err := os.WriteFile(logPath, content, 0o644); err != nil { + t.Fatalf("WriteFile failed: %v", err) + } + + stateStore := state.NewStore(filepath.Join(tempDir, "state.json")) + _, accessLogs, _ := BuildTrafficObservability(&config.Config{RouteConfigPath: routeConfigPath}, stateStore, nil) + if len(accessLogs) != 1 { + t.Fatalf("expected one access log, got %+v", accessLogs) + } + if got := len([]rune(accessLogs[0].Path)); got != accessLogPathMaxRunes { + t.Fatalf("expected truncated path length %d, got %d (%q)", accessLogPathMaxRunes, got, accessLogs[0].Path) + } +} + func TestBuildTrafficReportParsesCombinedAccessLog(t *testing.T) { tempDir := t.TempDir() routeConfigPath := filepath.Join(tempDir, "conf.d", "openflare_routes.conf") diff --git a/openflare_server/go.mod b/openflare_server/go.mod index 459f83f5..1b2f989e 100644 --- a/openflare_server/go.mod +++ b/openflare_server/go.mod @@ -28,6 +28,7 @@ require ( github.com/PuerkitoBio/purell v1.1.1 // indirect github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff // indirect + github.com/bwmarrin/snowflake v0.3.0 // indirect github.com/bytedance/sonic v1.11.2 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/chenzhuoyu/base64x v0.0.0-20230717121745-296ad89f973d // indirect @@ -58,6 +59,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/leodido/go-urn v1.4.0 // indirect + github.com/longbridgeapp/sqlparser v0.3.1 // indirect github.com/mailru/easyjson v0.7.6 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -67,6 +69,7 @@ require ( github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect golang.org/x/arch v0.7.0 // indirect + golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 // indirect golang.org/x/sync v0.18.0 // indirect golang.org/x/sys v0.38.0 // indirect golang.org/x/text v0.31.0 // indirect @@ -74,6 +77,7 @@ require ( google.golang.org/protobuf v1.33.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect + gorm.io/sharding v0.6.2 // indirect modernc.org/libc v1.22.5 // indirect modernc.org/mathutil v1.5.0 // indirect modernc.org/memory v1.5.0 // indirect diff --git a/openflare_server/go.sum b/openflare_server/go.sum index d97b8d63..e1428452 100644 --- a/openflare_server/go.sum +++ b/openflare_server/go.sum @@ -6,6 +6,8 @@ github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE= github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff h1:RmdPFa+slIr4SCBg4st/l/vZWVe9QJKMXGO60Bxbe04= github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff/go.mod h1:+RTT1BOk5P97fT2CiHkbFQwkK3mjsFAP6zCYV2aXtjw= +github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0= +github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE= github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM= github.com/bytedance/sonic v1.10.0-rc/go.mod h1:ElCzW+ufi8qKqNW0FY314xriJhyJhuoJ3gFZdAHF7NM= github.com/bytedance/sonic v1.11.2 h1:ywfwo0a/3j9HR8wsYGWsIWl2mvRsI950HyoxiBERw5A= @@ -76,6 +78,7 @@ github.com/go-playground/validator/v10 v10.19.0 h1:ol+5Fu+cSq9JD7SoSqe04GMI92cbn github.com/go-playground/validator/v10 v10.19.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM= github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI= github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo= +github.com/go-test/deep v1.0.7/go.mod h1:QV8Hv/iy04NyLBxAdO9njL0iVPN1S4d/A3NVv1V36o8= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw= @@ -126,6 +129,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/longbridgeapp/sqlparser v0.3.1 h1:iWOZWGIFgQrJRgobLXUNJdvqGRpbVXkyKUKUA5CNJBE= +github.com/longbridgeapp/sqlparser v0.3.1/go.mod h1:GIHaUq8zvYyHLCLMJJykx1CdM6LHtkUih/QaJXySSx4= github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= github.com/mailru/easyjson v0.7.6 h1:8yTIVnZgCoiM1TgqoeTl+LfU5Jg6/xL3QhGQnimLYnA= @@ -157,6 +162,7 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8= github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE= +github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= @@ -190,6 +196,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= +golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 h1:m64FZMko/V45gv0bNmrNYoDEq8U5YUhetc9cBWKS1TQ= +golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63/go.mod h1:0v4NqG35kSWCMzLaMeX+IQrlSnVE/bqGSyC2cz/9Le8= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= @@ -253,6 +261,8 @@ gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4= gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo= gorm.io/gorm v1.25.10 h1:dQpO+33KalOA+aFYGlK+EfxcI5MbO7EP2yYygwh9h+s= gorm.io/gorm v1.25.10/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= +gorm.io/sharding v0.6.2 h1:V9inmbdhN+RfWPEKTvbKKKv7qxLz1CneBDQvuL5P7jg= +gorm.io/sharding v0.6.2/go.mod h1:dXaAZv0qyUmLkLAciQ+NH2O1D1A4/ttrrZ/XK4xW9HU= modernc.org/libc v1.22.5 h1:91BNch/e5B0uPbJFgqbxXuOnxBQjlS//icfQEGmvyjE= modernc.org/libc v1.22.5/go.mod h1:jj+Z7dTNX8fBScMVNRAYZ/jF91K8fdT2hYMThc3YjBY= modernc.org/mathutil v1.5.0 h1:rV0Ko/6SfM+8G+yKiyI830l3Wuz1zRutdslNoQ0kfiQ= diff --git a/openflare_server/model/main.go b/openflare_server/model/main.go index a7767adc..d2308814 100644 --- a/openflare_server/model/main.go +++ b/openflare_server/model/main.go @@ -99,17 +99,13 @@ func CountTable(tableName string) (num int64) { func openDatabase() (*gorm.DB, string, error) { if common.SQLDSN != "" { - db, err := gorm.Open(postgres.Open(common.SQLDSN), &gorm.Config{ - PrepareStmt: true, - }) + db, err := gorm.Open(postgres.Open(common.SQLDSN), &gorm.Config{}) if err != nil { return nil, "", err } return db, "postgres", nil } - db, err := gorm.Open(sqlite.Open(common.SQLitePath), &gorm.Config{ - PrepareStmt: true, - }) + db, err := gorm.Open(sqlite.Open(common.SQLitePath), &gorm.Config{}) if err != nil { return nil, "", err } @@ -265,8 +261,17 @@ func migrateTableData(source *gorm.DB, target *gorm.DB, item dbModel) error { if batchLen == 0 { break } - if err := target.Create(batchPtr.Interface()).Error; err != nil { - return fmt.Errorf("write target table %s failed: %w", item.tableName, err) + if isShardedObservabilityTable(item.tableName) { + for index := 0; index < batchLen; index++ { + record := batchPtr.Elem().Index(index) + if err := target.Create(record.Addr().Interface()).Error; err != nil { + return fmt.Errorf("write target sharded table %s failed: %w", item.tableName, err) + } + } + } else { + if err := target.Create(batchPtr.Interface()).Error; err != nil { + return fmt.Errorf("write target table %s failed: %w", item.tableName, err) + } } migrated += int64(batchLen) offset += batchLen @@ -293,6 +298,9 @@ func InitDB() (err error) { os.Exit(1) } DB = db + if err = registerSharding(db, backend); err != nil { + return err + } if err = migrateProxyRouteEnableHTTPSColumn(db); err != nil { return err } diff --git a/openflare_server/model/main_test.go b/openflare_server/model/main_test.go index d98fe50b..396f934b 100644 --- a/openflare_server/model/main_test.go +++ b/openflare_server/model/main_test.go @@ -121,3 +121,36 @@ func TestMigrateTableDataCopiesRows(t *testing.T) { t.Fatalf("unexpected migrated option value: %s", gotOption.Value) } } + +func TestRegisterShardingAutoMigratesShardTables(t *testing.T) { + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "sharded.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite db: %v", err) + } + sqlDB, err := db.DB() + if err != nil { + t.Fatalf("get sql db: %v", err) + } + t.Cleanup(func() { + _ = sqlDB.Close() + }) + if err := registerSharding(db, "sqlite"); err != nil { + t.Fatalf("register sharding: %v", err) + } + if err := autoMigrateAll(db); err != nil { + t.Fatalf("auto migrate db: %v", err) + } + + for _, table := range []string{ + "node_metric_snapshots_00", + "node_metric_snapshots_09", + "node_request_reports_00", + "node_request_reports_09", + "node_access_logs_00", + "node_access_logs_09", + } { + if !db.Migrator().HasTable(table) { + t.Fatalf("expected sharded table %s to exist", table) + } + } +} diff --git a/openflare_server/model/node_access_log.go b/openflare_server/model/node_access_log.go index a52ffac6..9853a45a 100644 --- a/openflare_server/model/node_access_log.go +++ b/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) { diff --git a/openflare_server/model/node_metric_snapshot.go b/openflare_server/model/node_metric_snapshot.go index d106c0ad..1e956a12 100644 --- a/openflare_server/model/node_metric_snapshot.go +++ b/openflare_server/model/node_metric_snapshot.go @@ -1,6 +1,11 @@ package model -import "time" +import ( + "sort" + "time" + + "gorm.io/gorm" +) type NodeMetricSnapshot struct { ID uint `json:"id" gorm:"primaryKey"` @@ -39,10 +44,25 @@ func ListNodeMetricSnapshots(nodeID string, since time.Time, limit int) (snapsho } func ListMetricSnapshotsSince(since time.Time) (snapshots []*NodeMetricSnapshot, err error) { - query := DB.Order("captured_at desc") - if !since.IsZero() { - query = query.Where("captured_at >= ?", since) + rows, err := queryAcrossShards("node_metric_snapshots", func(tx *gorm.DB) ([]*NodeMetricSnapshot, error) { + var shardRows []*NodeMetricSnapshot + query := tx.Order("captured_at desc") + if !since.IsZero() { + query = query.Where("captured_at >= ?", since) + } + if err := query.Find(&shardRows).Error; err != nil { + return nil, err + } + return shardRows, nil + }) + if err != nil { + return nil, err } - err = query.Find(&snapshots).Error - return snapshots, err + sort.Slice(rows, func(i int, j int) bool { + if rows[i].CapturedAt.Equal(rows[j].CapturedAt) { + return rows[i].ID > rows[j].ID + } + return rows[i].CapturedAt.After(rows[j].CapturedAt) + }) + return rows, nil } diff --git a/openflare_server/model/node_request_report.go b/openflare_server/model/node_request_report.go index aa79e156..8fe5f2aa 100644 --- a/openflare_server/model/node_request_report.go +++ b/openflare_server/model/node_request_report.go @@ -1,6 +1,11 @@ package model -import "time" +import ( + "sort" + "time" + + "gorm.io/gorm" +) type NodeRequestReport struct { ID uint `json:"id" gorm:"primaryKey"` @@ -34,10 +39,25 @@ func ListNodeRequestReports(nodeID string, since time.Time, limit int) (reports } func ListRequestReportsSince(since time.Time) (reports []*NodeRequestReport, err error) { - query := DB.Order("window_ended_at desc") - if !since.IsZero() { - query = query.Where("window_ended_at >= ?", since) + rows, err := queryAcrossShards("node_request_reports", func(tx *gorm.DB) ([]*NodeRequestReport, error) { + var shardRows []*NodeRequestReport + query := tx.Order("window_ended_at desc") + if !since.IsZero() { + query = query.Where("window_ended_at >= ?", since) + } + if err := query.Find(&shardRows).Error; err != nil { + return nil, err + } + return shardRows, nil + }) + if err != nil { + return nil, err } - err = query.Find(&reports).Error - return reports, err + sort.Slice(rows, func(i int, j int) bool { + if rows[i].WindowEndedAt.Equal(rows[j].WindowEndedAt) { + return rows[i].ID > rows[j].ID + } + return rows[i].WindowEndedAt.After(rows[j].WindowEndedAt) + }) + return rows, nil } diff --git a/openflare_server/model/sharding.go b/openflare_server/model/sharding.go new file mode 100644 index 00000000..91548e54 --- /dev/null +++ b/openflare_server/model/sharding.go @@ -0,0 +1,81 @@ +package model + +import ( + "fmt" + "sort" + "strings" + + "gorm.io/gorm" + "gorm.io/sharding" +) + +const observabilityShardCount = 10 + +func registerSharding(db *gorm.DB, backend string) error { + if db == nil { + return nil + } + _ = backend + if err := db.Use(sharding.Register(sharding.Config{ + ShardingKey: "node_id", + NumberOfShards: observabilityShardCount, + PrimaryKeyGenerator: sharding.PKCustom, + PrimaryKeyGeneratorFn: func(tableIdx int64) int64 { + return 0 + }, + }, shardedObservabilityTables()...)); err != nil { + return fmt.Errorf("register observability sharding failed: %w", err) + } + return nil +} + +func shardedObservabilityTables() []any { + return []any{ + &NodeMetricSnapshot{}, + &NodeRequestReport{}, + &NodeAccessLog{}, + } +} + +func isShardedObservabilityTable(tableName string) bool { + switch strings.TrimSpace(tableName) { + case "node_metric_snapshots", "node_request_reports", "node_access_logs": + return true + default: + return false + } +} + +func observabilityShardTables(baseTable string) []string { + tables := make([]string, 0, observabilityShardCount) + for _, suffix := range observabilityShardSuffixes() { + tables = append(tables, baseTable+suffix) + } + return tables +} + +func observabilityShardSuffixes() []string { + suffixes := make([]string, 0, observabilityShardCount) + for index := 0; index < observabilityShardCount; index++ { + suffixes = append(suffixes, fmt.Sprintf("_%02d", index)) + } + return suffixes +} + +func queryAcrossShards[T any](baseTable string, query func(tx *gorm.DB) ([]T, error)) ([]T, error) { + items := make([]T, 0) + for _, table := range observabilityShardTables(baseTable) { + rows, err := query(DB.Table(table)) + if err != nil { + return nil, err + } + items = append(items, rows...) + } + return items, nil +} + +func sortShardRows[T any](items []T, less func(left T, right T) bool) { + sort.Slice(items, func(i int, j int) bool { + return less(items[i], items[j]) + }) +} diff --git a/openflare_server/service/access_log_test.go b/openflare_server/service/access_log_test.go index 7dc8d749..b9000415 100644 --- a/openflare_server/service/access_log_test.go +++ b/openflare_server/service/access_log_test.go @@ -2,6 +2,7 @@ package service import ( "openflare/model" + "strings" "testing" "time" ) @@ -60,9 +61,7 @@ func TestListAccessLogsIncludesSummaryTotals(t *testing.T) { StatusCode: 200, }, } - if err := model.DB.Create(&logs).Error; err != nil { - t.Fatalf("failed to seed access logs: %v", err) - } + seedNodeAccessLogs(t, logs) result, err := ListAccessLogs(AccessLogQuery{Page: 0, PageSize: 2}) if err != nil { @@ -121,9 +120,7 @@ func TestListAccessLogsUsesDefaultPageSize(t *testing.T) { StatusCode: 200, }) } - if err := model.DB.Create(&logs).Error; err != nil { - t.Fatalf("failed to seed access logs: %v", err) - } + seedNodeAccessLogs(t, logs) result, err := ListAccessLogs(AccessLogQuery{}) if err != nil { @@ -143,7 +140,7 @@ func TestListAccessLogsUsesDefaultPageSize(t *testing.T) { func TestListFoldedAccessLogsAndIPSummaries(t *testing.T) { setupServiceTestDB(t) - now := time.Now().UTC() + now := time.Date(2026, 3, 19, 8, 12, 30, 0, time.UTC) if err := model.DB.Create(&model.Node{ NodeID: "node-folded", Name: "edge-folded", @@ -176,9 +173,7 @@ func TestListFoldedAccessLogsAndIPSummaries(t *testing.T) { StatusCode: 404, }, } - if err := model.DB.Create(&logs).Error; err != nil { - t.Fatalf("failed to seed access logs: %v", err) - } + seedNodeAccessLogs(t, logs) folded, err := ListFoldedAccessLogs(AccessLogQuery{ NodeID: "node-folded", @@ -221,7 +216,7 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) { setupServiceTestDB(t) now := time.Now().UTC() - if err := model.DB.Create([]*model.NodeAccessLog{ + seedNodeAccessLogs(t, []*model.NodeAccessLog{ { NodeID: "node-cleanup", LoggedAt: now.Add(-10 * 24 * time.Hour), @@ -238,9 +233,7 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) { Path: "/recent", StatusCode: 200, }, - }).Error; err != nil { - t.Fatalf("failed to seed cleanup logs: %v", err) - } + }) result, err := CleanupAccessLogs(AccessLogCleanupInput{RetentionDays: 7}) if err != nil { @@ -258,3 +251,45 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) { t.Fatalf("unexpected remaining logs after cleanup: %+v", remaining.Items) } } + +func TestPersistNodeAccessLogsTruncatesLongPath(t *testing.T) { + setupServiceTestDB(t) + + longPath := "/" + strings.Repeat("a", 140) + reportedAt := time.Now().UTC() + if err := persistNodeAccessLogs(model.DB, "node-truncate", []AgentNodeAccessLog{ + { + LoggedAtUnix: reportedAt.Unix(), + RemoteAddr: "203.0.113.10", + Host: "truncate.example.com", + Path: longPath, + StatusCode: 200, + }, + }, reportedAt); err != nil { + t.Fatalf("persistNodeAccessLogs failed: %v", err) + } + + logs, err := model.ListNodeAccessLogs(model.NodeAccessLogQuery{ + NodeID: "node-truncate", + Page: 0, + PageSize: 10, + }) + if err != nil { + t.Fatalf("ListNodeAccessLogs failed: %v", err) + } + if len(logs) != 1 { + t.Fatalf("expected one stored log, got %+v", logs) + } + if got := len([]rune(logs[0].Path)); got != nodeAccessLogPathMaxLength { + t.Fatalf("expected truncated path length %d, got %d (%q)", nodeAccessLogPathMaxLength, got, logs[0].Path) + } +} + +func seedNodeAccessLogs(t *testing.T, logs []*model.NodeAccessLog) { + t.Helper() + for _, item := range logs { + if err := model.DB.Create(item).Error; err != nil { + t.Fatalf("failed to seed access log: %v", err) + } + } +} diff --git a/openflare_server/service/node_update_test.go b/openflare_server/service/node_update_test.go index b4348ce6..3d5d1e58 100644 --- a/openflare_server/service/node_update_test.go +++ b/openflare_server/service/node_update_test.go @@ -908,7 +908,7 @@ func TestListAccessLogsUsesPagination(t *testing.T) { } now := time.Now().UTC() - if err := model.DB.Create([]*model.NodeAccessLog{ + for _, item := range []*model.NodeAccessLog{ { NodeID: node.NodeID, LoggedAt: now.Add(-10 * time.Second), @@ -936,8 +936,10 @@ func TestListAccessLogsUsesPagination(t *testing.T) { Path: "/three", StatusCode: 502, }, - }).Error; err != nil { - t.Fatalf("failed to seed access logs: %v", err) + } { + if err := model.DB.Create(item).Error; err != nil { + t.Fatalf("failed to seed access logs: %v", err) + } } pageOne, err := ListAccessLogs(AccessLogQuery{ @@ -1404,7 +1406,7 @@ func TestGetDashboardOverview(t *testing.T) { }).Insert(); err != nil { t.Fatalf("failed to insert node b traffic report: %v", err) } - if err := model.DB.Create([]*model.NodeAccessLog{ + for _, item := range []*model.NodeAccessLog{ { NodeID: "node-dashboard-a", LoggedAt: now.Add(-30 * time.Minute), @@ -1432,8 +1434,10 @@ func TestGetDashboardOverview(t *testing.T) { Path: "/edge", StatusCode: 502, }, - }).Error; err != nil { - t.Fatalf("failed to seed dashboard access logs: %v", err) + } { + if err := model.DB.Create(item).Error; err != nil { + t.Fatalf("failed to seed dashboard access logs: %v", err) + } } if err := model.DB.Create(&model.NodeHealthEvent{ diff --git a/openflare_server/service/observability.go b/openflare_server/service/observability.go index 73b8f622..84cb10ff 100644 --- a/openflare_server/service/observability.go +++ b/openflare_server/service/observability.go @@ -18,6 +18,7 @@ const ( NodeHealthSeverityWarning = "warning" NodeHealthSeverityCritical = "critical" nodeAccessLogRetentionWindow = nodeAccessLogRetentionDays * 24 * time.Hour + nodeAccessLogPathMaxLength = 100 ) type AgentNodeSystemProfile struct { @@ -177,7 +178,7 @@ func persistNodeMetricSnapshot(tx *gorm.DB, nodeID string, snapshot *AgentNodeMe OpenrestyConnections: snapshot.OpenrestyConnections, RawJSON: marshalJSON(snapshot), } - return tx.Where("node_id = ? AND captured_at = ?", nodeID, record.CapturedAt).Assign(record).FirstOrCreate(record).Error + return tx.Where("node_id = ? AND captured_at = ?", nodeID, record.CapturedAt).FirstOrCreate(record).Error } func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *AgentNodeTrafficReport, reportedAt time.Time) error { @@ -199,7 +200,7 @@ func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *AgentNodeTraff SourceCountriesJSON: marshalJSON(report.SourceCountries), RawJSON: marshalJSON(report), } - return tx.Where("node_id = ? AND window_started_at = ? AND window_ended_at = ?", nodeID, record.WindowStartedAt, record.WindowEndedAt).Assign(record).FirstOrCreate(record).Error + return tx.Where("node_id = ? AND window_started_at = ? AND window_ended_at = ?", nodeID, record.WindowStartedAt, record.WindowEndedAt).FirstOrCreate(record).Error } func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog, reportedAt time.Time) error { @@ -220,7 +221,7 @@ func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog RemoteAddr: strings.TrimSpace(item.RemoteAddr), Region: "", Host: strings.TrimSpace(item.Host), - Path: strings.TrimSpace(item.Path), + Path: truncateForDatabase(strings.TrimSpace(item.Path), nodeAccessLogPathMaxLength), StatusCode: item.StatusCode, RawJSON: marshalJSON(item), } @@ -235,7 +236,7 @@ func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog record.Host, record.Path, record.StatusCode, - ).Assign(record).FirstOrCreate(record).Error; err != nil { + ).FirstOrCreate(record).Error; err != nil { return err } }