diff --git a/Makefile b/Makefile index e272843..d774bdf 100644 --- a/Makefile +++ b/Makefile @@ -79,7 +79,7 @@ build-legacy: # Run the application (new CLI) run: docker-up @echo "Waiting for Neo4j to be ready..." - @sleep 15 + @sleep 30 @echo "Starting application..." go run cmd/sql-graph-visualizer/main.go serve diff --git a/cmd/main.go b/cmd/main.go index 8b78569..23857c6 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -274,6 +274,7 @@ func main() { performanceServices.GraphMapper, performanceServices.RealtimeMonitor, performanceServices.PSAdapter, + neo4jRepo, ) performanceHandlers.RegisterRoutes(router) logrus.Info("Performance API routes registered") diff --git a/go.mod b/go.mod index 66c37bd..ce7d6bd 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ toolchain go1.26.4 require ( github.com/99designs/gqlgen v0.17.91 - github.com/go-sql-driver/mysql v1.9.3 + github.com/go-sql-driver/mysql v1.10.0 github.com/google/uuid v1.6.0 github.com/gorilla/mux v1.8.1 github.com/gorilla/websocket v1.5.3 @@ -19,7 +19,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 github.com/vektah/gqlparser/v2 v2.5.34 - golang.org/x/text v0.38.0 + golang.org/x/text v0.39.0 gopkg.in/yaml.v2 v2.4.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -27,7 +27,7 @@ require ( require ( filippo.io/edwards25519 v1.2.0 // indirect github.com/agnivade/levenshtein v1.2.1 // indirect - github.com/coder/websocket v1.8.14 // indirect + github.com/coder/websocket v1.8.15 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect github.com/goccy/go-yaml v1.19.2 // indirect @@ -42,9 +42,9 @@ require ( github.com/spf13/pflag v1.0.10 // indirect github.com/stretchr/objx v0.5.3 // indirect github.com/urfave/cli/v3 v3.10.0 // indirect - golang.org/x/crypto v0.50.0 // indirect + golang.org/x/crypto v0.53.0 // indirect golang.org/x/mod v0.37.0 // indirect golang.org/x/sync v0.21.0 // indirect golang.org/x/sys v0.46.0 // indirect - golang.org/x/tools v0.46.0 // indirect + golang.org/x/tools v0.47.0 // indirect ) diff --git a/go.sum b/go.sum index 1faa3fd..0e5832f 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,8 @@ github.com/agnivade/levenshtein v1.2.1 h1:EHBY3UOn1gwdy/VbFwgo4cxecRznFk7fKWN1KO github.com/agnivade/levenshtein v1.2.1/go.mod h1:QVVI16kDrtSuwcpd0p1+xMC6Z/VfhtCyDIjcwga4/DU= github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0 h1:jfIu9sQUG6Ig+0+Ap1h4unLjW6YQJpKZVmUzxsD4E/Q= github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0/go.mod h1:t2tdKJDJF9BV14lnkjHmOQgcvEKgtqs5a1N3LNdJhGE= -github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= -github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -31,8 +31,8 @@ github.com/dgryski/trifles v0.0.0-20230903005119-f50d829f2e54/go.mod h1:if7Fbed8 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU= -github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= -github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= +github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw= +github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk= github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= @@ -130,8 +130,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= -golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= +golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= @@ -168,13 +168,13 @@ golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9sn golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= -golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= +golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= +golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.46.0 h1:7jTurBkPZu4moS/Uy4OQT1M+QBlsj3wejyZwsT8Z7rk= -golang.org/x/tools v0.46.0/go.mod h1:FrD85F8l+NWL+9XWBSyVSHO6Ne4jutsfIFba7AWQ5Ys= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/internal/application/bootstrap/servers.go b/internal/application/bootstrap/servers.go index 60c0773..f47e783 100644 --- a/internal/application/bootstrap/servers.go +++ b/internal/application/bootstrap/servers.go @@ -21,6 +21,7 @@ import ( "github.com/gorilla/mux" "github.com/sirupsen/logrus" + "sql-graph-visualizer/internal/application/ports" graphqlserver "sql-graph-visualizer/internal/application/services/graphql" "sql-graph-visualizer/internal/application/services/performance" "sql-graph-visualizer/internal/domain/aggregates/graph" @@ -179,6 +180,7 @@ func (r *Resources) startAPIServer() (*http.Server, error) { r.PerformanceServices.GraphMapper, r.PerformanceServices.RealtimeMonitor, r.PerformanceServices.PSAdapter, + r.Neo4jRepo, ) performanceHandlers.RegisterRoutes(router) logrus.Info("Performance API routes registered") @@ -361,10 +363,17 @@ func initPerformanceServices(cfg *models.Config, db *sql.DB) *PerformanceService benchmarkConfig := createBenchmarkConfig(cfg) // The source/graph repositories are intentionally nil here: the current // benchmark execution path runs sysbench, which connects to the database - // directly via the configured DatabaseURL. Repositories will be wired in when - // benchmark-result persistence is added. + + // directly via the configured DatabaseURL. benchmarkService := performance.NewBenchmarkService(nil, nil, nil, performanceAnalyzer, logger, benchmarkConfig) - registerBenchmarkTools(benchmarkService, cfg, logger) + registerBenchmarkTools(benchmarkService, cfg, db, logger) + + if store, err := performance.NewFileBenchmarkResultStore(logger, benchmarkResultsDir(cfg)); err != nil { + logrus.Warnf("Benchmark result persistence unavailable, falling back to in-memory results only: %v", err) + } else { + benchmarkService.SetResultStore(store) + } + if cfg.Performance.Realtime != nil && cfg.Performance.Realtime.Enabled { ctx := context.Background() @@ -385,20 +394,28 @@ func initPerformanceServices(cfg *models.Config, db *sql.DB) *PerformanceService } func createGraphMapperConfig(cfg *models.Config) *performance.GraphPerformanceMapperConfig { - config := &performance.GraphPerformanceMapperConfig{} + // Start from sane defaults so an invalid or empty duration in the user + // configuration falls back to a working value instead of silently + // zeroing the interval (which would otherwise busy-loop or never update). + config := performance.DefaultGraphPerformanceMapperConfig() if cfg.Performance.Visualization != nil { - updateInterval, _ := time.ParseDuration(cfg.Performance.Visualization.UpdateInterval) - historyRetention, _ := time.ParseDuration(cfg.Performance.Visualization.HistoryRetention) - config.UpdateInterval = updateInterval - config.HistoryRetention = historyRetention - config.MaxConcurrentUpdates = cfg.Performance.Visualization.MaxConcurrentUpdates - if cfg.Performance.Visualization.EdgeThickness != nil { + v := cfg.Performance.Visualization + if d, err := time.ParseDuration(v.UpdateInterval); err == nil && d > 0 { + config.UpdateInterval = d + } + if d, err := time.ParseDuration(v.HistoryRetention); err == nil && d > 0 { + config.HistoryRetention = d + } + if v.MaxConcurrentUpdates > 0 { + config.MaxConcurrentUpdates = v.MaxConcurrentUpdates + } + if v.EdgeThickness != nil { config.EdgeThickness = performance.EdgeThicknessConfig{ - Metric: cfg.Performance.Visualization.EdgeThickness.Metric, - Scale: cfg.Performance.Visualization.EdgeThickness.Scale, - MinThickness: cfg.Performance.Visualization.EdgeThickness.MinThickness, - MaxThickness: cfg.Performance.Visualization.EdgeThickness.MaxThickness, - Multiplier: cfg.Performance.Visualization.EdgeThickness.Multiplier, + Metric: v.EdgeThickness.Metric, + Scale: v.EdgeThickness.Scale, + MinThickness: v.EdgeThickness.MinThickness, + MaxThickness: v.EdgeThickness.MaxThickness, + Multiplier: v.EdgeThickness.Multiplier, } } } @@ -406,29 +423,41 @@ func createGraphMapperConfig(cfg *models.Config) *performance.GraphPerformanceMa } func createRealtimeConfig(cfg *models.Config) *performance.RealtimeMonitorConfig { - config := &performance.RealtimeMonitorConfig{} + // Start from sane defaults; only override fields the user actually set so + // an invalid/empty duration doesn't silently zero the interval. + config := performance.DefaultRealtimeMonitorConfig() if cfg.Performance.Realtime != nil { - updateInterval, _ := time.ParseDuration(cfg.Performance.Realtime.UpdateInterval) - heartbeatInterval, _ := time.ParseDuration(cfg.Performance.Realtime.HeartbeatInterval) - writeTimeout, _ := time.ParseDuration(cfg.Performance.Realtime.WriteTimeout) - readTimeout, _ := time.ParseDuration(cfg.Performance.Realtime.ReadTimeout) - pingTimeout, _ := time.ParseDuration(cfg.Performance.Realtime.PingTimeout) - config.DataUpdateInterval = updateInterval - config.HeartbeatInterval = heartbeatInterval - config.MaxConnections = cfg.Performance.Realtime.MaxConnections - config.WriteTimeout = writeTimeout - config.ReadTimeout = readTimeout - config.PingTimeout = pingTimeout - config.MaxMessageSize = cfg.Performance.Realtime.MaxMessageSize - config.CompressionEnabled = cfg.Performance.Realtime.CompressionEnabled - if cfg.Performance.Realtime.Alerts != nil { + r := cfg.Performance.Realtime + if d, err := time.ParseDuration(r.UpdateInterval); err == nil && d > 0 { + config.DataUpdateInterval = d + } + if d, err := time.ParseDuration(r.HeartbeatInterval); err == nil && d > 0 { + config.HeartbeatInterval = d + } + if d, err := time.ParseDuration(r.WriteTimeout); err == nil && d > 0 { + config.WriteTimeout = d + } + if d, err := time.ParseDuration(r.ReadTimeout); err == nil && d > 0 { + config.ReadTimeout = d + } + if d, err := time.ParseDuration(r.PingTimeout); err == nil && d > 0 { + config.PingTimeout = d + } + if r.MaxConnections > 0 { + config.MaxConnections = r.MaxConnections + } + if r.MaxMessageSize > 0 { + config.MaxMessageSize = r.MaxMessageSize + } + config.CompressionEnabled = r.CompressionEnabled + if r.Alerts != nil { config.AlertThresholds = performance.AlertThresholds{ - HighLatency: cfg.Performance.Realtime.Alerts.HighLatency, - HighErrorRate: cfg.Performance.Realtime.Alerts.HighErrorRate, - HighCPUUsage: cfg.Performance.Realtime.Alerts.HighCPUUsage, - HighMemoryUsage: cfg.Performance.Realtime.Alerts.HighMemoryUsage, - SlowQueryThreshold: cfg.Performance.Realtime.Alerts.SlowQueryThreshold, - DeadlockThreshold: cfg.Performance.Realtime.Alerts.DeadlockThreshold, + HighLatency: r.Alerts.HighLatency, + HighErrorRate: r.Alerts.HighErrorRate, + HighCPUUsage: r.Alerts.HighCPUUsage, + HighMemoryUsage: r.Alerts.HighMemoryUsage, + SlowQueryThreshold: r.Alerts.SlowQueryThreshold, + DeadlockThreshold: r.Alerts.DeadlockThreshold, } } } @@ -489,10 +518,11 @@ func createBenchmarkConfig(cfg *models.Config) *performance.BenchmarkServiceConf return config } -// registerBenchmarkTools wires the available benchmark tools (currently -// sysbench) into the benchmark service. Unavailable tools are logged and -// skipped so the application keeps running without benchmarking support. -func registerBenchmarkTools(svc *performance.BenchmarkService, cfg *models.Config, logger *logrus.Logger) { +// registerBenchmarkTools wires the available benchmark tools (sysbench and, +// when configured, custom query sets) into the benchmark service. +// Unavailable tools are logged and skipped so the application keeps running +// without benchmarking support. +func registerBenchmarkTools(svc *performance.BenchmarkService, cfg *models.Config, db *sql.DB, logger *logrus.Logger) { if b := cfg.Performance.Benchmarks; b != nil && !b.Enabled { logrus.Info("Benchmarking disabled in configuration; skipping benchmark tool registration") return @@ -501,9 +531,66 @@ func registerBenchmarkTools(svc *performance.BenchmarkService, cfg *models.Confi sysbenchAdapter := performance.NewSysbenchAdapter(logger, createSysbenchConfig(cfg)) if err := svc.RegisterBenchmarkTool("sysbench", sysbenchAdapter); err != nil { logrus.Warnf("Sysbench benchmark tool unavailable; sysbench benchmarks disabled: %v", err) + } else { + logrus.Info("Registered sysbench benchmark tool") + } + + querySets := createCustomQuerySets(cfg) + if len(querySets) == 0 { + return + } + customAdapter := performance.NewCustomQueryAdapter(logger, db, querySets) + if err := svc.RegisterBenchmarkTool("custom", customAdapter); err != nil { + logrus.Warnf("Custom query benchmark tool unavailable: %v", err) return } - logrus.Info("Registered sysbench benchmark tool") + logrus.Infof("Registered custom query benchmark tool with %d query set(s)", len(querySets)) +} + +// createCustomQuerySets converts the user-configured custom query benchmark +// sets into the ports representation consumed by CustomQueryAdapter. +func createCustomQuerySets(cfg *models.Config) map[string]ports.CustomBenchmarkConfig { + result := make(map[string]ports.CustomBenchmarkConfig) + b := cfg.Performance.Benchmarks + if b == nil { + return result + } + + for _, set := range b.CustomQueries { + if set.Name == "" || len(set.Queries) == 0 { + continue + } + + duration, _ := time.ParseDuration(set.Duration) + + queries := make([]ports.CustomQueryDefinition, 0, len(set.Queries)) + for _, q := range set.Queries { + if q.Query == "" { + continue + } + expectedLatency, _ := time.ParseDuration(q.ExpectedLatency) + queries = append(queries, ports.CustomQueryDefinition{ + Query: q.Query, + Weight: q.Weight, + Parameters: q.Parameters, + Description: q.Description, + ExpectedLatency: expectedLatency, + TargetQPS: q.TargetQPS, + }) + } + if len(queries) == 0 { + continue + } + + result[set.Name] = ports.CustomBenchmarkConfig{ + Name: set.Name, + Description: set.Description, + Duration: duration, + Threads: set.Threads, + Queries: queries, + } + } + return result } // createSysbenchConfig maps user configuration onto the sysbench adapter config, @@ -523,6 +610,15 @@ func createSysbenchConfig(cfg *models.Config) *performance.SysbenchConfig { return sbConfig } +// benchmarkResultsDir returns the configured directory for persisted +// benchmark results, falling back to a sane default when unset. +func benchmarkResultsDir(cfg *models.Config) string { + if b := cfg.Performance.Benchmarks; b != nil && b.ResultsDir != "" { + return b.ResultsDir + } + return "data/performance/benchmarks" +} + // benchmarkDatabaseURL builds a sysbench-compatible database URL and driver type // from the active source database configuration. func benchmarkDatabaseURL(cfg *models.Config) (string, string) { diff --git a/internal/application/ports/benchmark_result_store_port.go b/internal/application/ports/benchmark_result_store_port.go new file mode 100644 index 0000000..fb1c361 --- /dev/null +++ b/internal/application/ports/benchmark_result_store_port.go @@ -0,0 +1,39 @@ +// Package ports defines interfaces for application layer dependencies. +package ports + +import ( + "context" + "time" +) + +// BenchmarkResultStorePort defines persistence for historical benchmark +// results, allowing trend analysis and regression detection to survive +// process restarts. +type BenchmarkResultStorePort interface { + // Save persists a completed (or failed/cancelled) benchmark result. + Save(ctx context.Context, result *BenchmarkResult) error + + // Get retrieves a single stored result by its execution ID. + Get(ctx context.Context, id string) (*BenchmarkResult, error) + + // List returns stored results matching the given filter, ordered by + // start time ascending. + List(ctx context.Context, filter BenchmarkResultFilter) ([]*BenchmarkResult, error) + + // DeleteOlderThan removes stored results whose start time is before the + // given cutoff. It returns the number of deleted results. + DeleteOlderThan(ctx context.Context, cutoff time.Time) (int, error) +} + +// BenchmarkResultFilter narrows down which stored benchmark results to +// return from List. Zero-valued fields are treated as "no filter" for that +// dimension. +type BenchmarkResultFilter struct { + ToolName string + TestType string + Since time.Time + Until time.Time + // Limit caps the number of results returned (most recent first). Zero + // or negative means no limit. + Limit int +} diff --git a/internal/application/services/performance/benchmark_result_store.go b/internal/application/services/performance/benchmark_result_store.go new file mode 100644 index 0000000..dcdaf48 --- /dev/null +++ b/internal/application/services/performance/benchmark_result_store.go @@ -0,0 +1,239 @@ +package performance + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "sync" + "time" + + "sql-graph-visualizer/internal/application/ports" + + "github.com/sirupsen/logrus" +) + +// FileBenchmarkResultStore is a simple, dependency-free implementation of +// ports.BenchmarkResultStorePort that appends benchmark results as +// newline-delimited JSON (JSONL) to a file on disk. All stored results are +// also kept in an in-memory cache so reads never need to re-parse the file. +// +// This is intentionally simple: benchmark executions are infrequent +// (human-triggered or scheduled), so a single append-only file with an +// in-memory index is sufficient and avoids introducing a new database +// dependency (e.g. SQLite) purely for this feature. +type FileBenchmarkResultStore struct { + logger *logrus.Logger + dir string + filePath string + + mu sync.Mutex + results map[string]*ports.BenchmarkResult +} + +// NewFileBenchmarkResultStore creates a store rooted at dir, creating the +// directory if necessary and loading any previously persisted results into +// memory. +func NewFileBenchmarkResultStore(logger *logrus.Logger, dir string) (*FileBenchmarkResultStore, error) { + if dir == "" { + dir = "data/performance/benchmarks" + } + if err := os.MkdirAll(dir, 0o750); err != nil { + return nil, fmt.Errorf("failed to create benchmark results directory %q: %w", dir, err) + } + + store := &FileBenchmarkResultStore{ + logger: logger, + dir: dir, + filePath: filepath.Join(dir, "benchmark_results.jsonl"), + results: make(map[string]*ports.BenchmarkResult), + } + + if err := store.load(); err != nil { + return nil, err + } + + return store, nil +} + +// load reads all previously persisted results from the JSONL file into the +// in-memory cache. Malformed lines are logged and skipped rather than +// aborting startup. +func (s *FileBenchmarkResultStore) load() error { + f, err := os.Open(s.filePath) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("failed to open benchmark results file %q: %w", s.filePath, err) + } + defer func() { _ = f.Close() }() + + scanner := bufio.NewScanner(f) + // Benchmark results can carry a sizeable number of query results; allow + // lines larger than bufio's 64KB default. + scanner.Buffer(make([]byte, 0, 64*1024), 10*1024*1024) + + loaded := 0 + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + var result ports.BenchmarkResult + if err := json.Unmarshal(line, &result); err != nil { + s.logger.WithError(err).Warn("Skipping malformed benchmark result record") + continue + } + s.results[result.ID] = &result + loaded++ + } + if err := scanner.Err(); err != nil { + return fmt.Errorf("failed to read benchmark results file %q: %w", s.filePath, err) + } + + s.logger.WithFields(logrus.Fields{"count": loaded, "path": s.filePath}).Info("Loaded persisted benchmark results") + return nil +} + +// Save persists a benchmark result, appending it to the JSONL file and +// updating the in-memory cache. +func (s *FileBenchmarkResultStore) Save(_ context.Context, result *ports.BenchmarkResult) error { + if result == nil { + return fmt.Errorf("cannot save a nil benchmark result") + } + + data, err := json.Marshal(result) + if err != nil { + return fmt.Errorf("failed to marshal benchmark result: %w", err) + } + + s.mu.Lock() + defer s.mu.Unlock() + + f, err := os.OpenFile(s.filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o640) + if err != nil { + return fmt.Errorf("failed to open benchmark results file %q: %w", s.filePath, err) + } + defer func() { _ = f.Close() }() + + if _, err := f.Write(append(data, '\n')); err != nil { + return fmt.Errorf("failed to append benchmark result: %w", err) + } + + resultCopy := *result + s.results[result.ID] = &resultCopy + return nil +} + +// Get retrieves a single stored result by ID. +func (s *FileBenchmarkResultStore) Get(_ context.Context, id string) (*ports.BenchmarkResult, error) { + s.mu.Lock() + defer s.mu.Unlock() + + result, exists := s.results[id] + if !exists { + return nil, fmt.Errorf("benchmark result %q not found", id) + } + resultCopy := *result + return &resultCopy, nil +} + +// List returns stored results matching the filter, ordered by start time +// ascending (oldest first). When filter.Limit > 0, only the most recent +// matching results (up to the limit) are returned, still ordered ascending. +func (s *FileBenchmarkResultStore) List(_ context.Context, filter ports.BenchmarkResultFilter) ([]*ports.BenchmarkResult, error) { + s.mu.Lock() + matches := make([]*ports.BenchmarkResult, 0, len(s.results)) + for _, result := range s.results { + if filter.ToolName != "" && result.ToolName != filter.ToolName { + continue + } + if filter.TestType != "" && result.TestType != filter.TestType { + continue + } + if !filter.Since.IsZero() && result.StartTime.Before(filter.Since) { + continue + } + if !filter.Until.IsZero() && result.StartTime.After(filter.Until) { + continue + } + resultCopy := *result + matches = append(matches, &resultCopy) + } + s.mu.Unlock() + + sort.Slice(matches, func(i, j int) bool { + return matches[i].StartTime.Before(matches[j].StartTime) + }) + + if filter.Limit > 0 && len(matches) > filter.Limit { + matches = matches[len(matches)-filter.Limit:] + } + + return matches, nil +} + +// DeleteOlderThan removes results started before cutoff, compacting the +// backing file to reflect the remaining set. +func (s *FileBenchmarkResultStore) DeleteOlderThan(_ context.Context, cutoff time.Time) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + deleted := 0 + for id, result := range s.results { + if result.StartTime.Before(cutoff) { + delete(s.results, id) + deleted++ + } + } + + if deleted == 0 { + return 0, nil + } + + if err := s.rewriteLocked(); err != nil { + return 0, err + } + + s.logger.WithField("deleted_count", deleted).Info("Compacted persisted benchmark results") + return deleted, nil +} + +// rewriteLocked rewrites the backing file from the current in-memory cache. +// Callers must hold s.mu. +func (s *FileBenchmarkResultStore) rewriteLocked() error { + tmpPath := s.filePath + ".tmp" + f, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o640) + if err != nil { + return fmt.Errorf("failed to create temporary benchmark results file: %w", err) + } + + writer := bufio.NewWriter(f) + for _, result := range s.results { + data, err := json.Marshal(result) + if err != nil { + _ = f.Close() + return fmt.Errorf("failed to marshal benchmark result during compaction: %w", err) + } + if _, err := writer.Write(append(data, '\n')); err != nil { + _ = f.Close() + return fmt.Errorf("failed to write benchmark result during compaction: %w", err) + } + } + if err := writer.Flush(); err != nil { + _ = f.Close() + return fmt.Errorf("failed to flush compacted benchmark results file: %w", err) + } + if err := f.Close(); err != nil { + return fmt.Errorf("failed to close compacted benchmark results file: %w", err) + } + + if err := os.Rename(tmpPath, s.filePath); err != nil { + return fmt.Errorf("failed to replace benchmark results file: %w", err) + } + return nil +} diff --git a/internal/application/services/performance/benchmark_result_store_test.go b/internal/application/services/performance/benchmark_result_store_test.go new file mode 100644 index 0000000..3f004f2 --- /dev/null +++ b/internal/application/services/performance/benchmark_result_store_test.go @@ -0,0 +1,174 @@ +package performance + +import ( + "context" + "path/filepath" + "testing" + "time" + + "sql-graph-visualizer/internal/application/ports" + + "github.com/sirupsen/logrus" +) + +func newTestLogger() *logrus.Logger { + logger := logrus.New() + logger.SetLevel(logrus.PanicLevel) + return logger +} + +func TestFileBenchmarkResultStore_SaveGetList(t *testing.T) { + dir := t.TempDir() + store, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("NewFileBenchmarkResultStore() error = %v", err) + } + + ctx := context.Background() + now := time.Now() + + results := []*ports.BenchmarkResult{ + {ID: "run-1", ToolName: "sysbench", TestType: "oltp_read_write", StartTime: now.Add(-2 * time.Hour), EndTime: now.Add(-2 * time.Hour), Status: ports.BenchmarkStatusCompleted}, + {ID: "run-2", ToolName: "sysbench", TestType: "oltp_read_only", StartTime: now.Add(-1 * time.Hour), EndTime: now.Add(-1 * time.Hour), Status: ports.BenchmarkStatusCompleted}, + {ID: "run-3", ToolName: "custom", TestType: "reporting_queries", StartTime: now, EndTime: now, Status: ports.BenchmarkStatusFailed}, + } + for _, r := range results { + if err := store.Save(ctx, r); err != nil { + t.Fatalf("Save(%s) error = %v", r.ID, err) + } + } + + got, err := store.Get(ctx, "run-2") + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if got.ID != "run-2" || got.TestType != "oltp_read_only" { + t.Errorf("Get() = %+v, want run-2/oltp_read_only", got) + } + + if _, err := store.Get(ctx, "missing"); err == nil { + t.Error("Get() for missing ID: expected error, got nil") + } + + all, err := store.List(ctx, ports.BenchmarkResultFilter{}) + if err != nil { + t.Fatalf("List() error = %v", err) + } + if len(all) != 3 { + t.Fatalf("List() returned %d results, want 3", len(all)) + } + // Must be ordered ascending by start time. + for i := 1; i < len(all); i++ { + if all[i-1].StartTime.After(all[i].StartTime) { + t.Errorf("List() not sorted ascending by start time: %v before %v", all[i-1].StartTime, all[i].StartTime) + } + } + + sysbenchOnly, err := store.List(ctx, ports.BenchmarkResultFilter{ToolName: "sysbench"}) + if err != nil { + t.Fatalf("List(tool filter) error = %v", err) + } + if len(sysbenchOnly) != 2 { + t.Errorf("List(tool=sysbench) returned %d results, want 2", len(sysbenchOnly)) + } + + limited, err := store.List(ctx, ports.BenchmarkResultFilter{Limit: 1}) + if err != nil { + t.Fatalf("List(limit) error = %v", err) + } + if len(limited) != 1 || limited[0].ID != "run-3" { + t.Errorf("List(limit=1) = %+v, want the single most recent result (run-3)", limited) + } +} + +func TestFileBenchmarkResultStore_PersistsAcrossInstances(t *testing.T) { + dir := t.TempDir() + ctx := context.Background() + + store1, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("NewFileBenchmarkResultStore() error = %v", err) + } + if err := store1.Save(ctx, &ports.BenchmarkResult{ID: "persisted-1", ToolName: "sysbench", StartTime: time.Now()}); err != nil { + t.Fatalf("Save() error = %v", err) + } + + store2, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("second NewFileBenchmarkResultStore() error = %v", err) + } + got, err := store2.Get(ctx, "persisted-1") + if err != nil { + t.Fatalf("Get() after reload error = %v", err) + } + if got.ID != "persisted-1" { + t.Errorf("Get() after reload = %+v, want persisted-1", got) + } + + if _, err := filepath.Abs(dir); err != nil { + t.Fatalf("filepath.Abs() error = %v", err) + } +} + +func TestFileBenchmarkResultStore_DeleteOlderThan(t *testing.T) { + dir := t.TempDir() + store, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("NewFileBenchmarkResultStore() error = %v", err) + } + + ctx := context.Background() + now := time.Now() + old := &ports.BenchmarkResult{ID: "old", StartTime: now.Add(-48 * time.Hour)} + recent := &ports.BenchmarkResult{ID: "recent", StartTime: now} + + if err := store.Save(ctx, old); err != nil { + t.Fatalf("Save(old) error = %v", err) + } + if err := store.Save(ctx, recent); err != nil { + t.Fatalf("Save(recent) error = %v", err) + } + + deleted, err := store.DeleteOlderThan(ctx, now.Add(-24*time.Hour)) + if err != nil { + t.Fatalf("DeleteOlderThan() error = %v", err) + } + if deleted != 1 { + t.Errorf("DeleteOlderThan() deleted = %d, want 1", deleted) + } + + if _, err := store.Get(ctx, "old"); err == nil { + t.Error("Get(old) after DeleteOlderThan: expected error, got nil") + } + if _, err := store.Get(ctx, "recent"); err != nil { + t.Errorf("Get(recent) after DeleteOlderThan: unexpected error %v", err) + } + + // Reload from disk to confirm compaction persisted correctly. + reloaded, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("reload after compaction error = %v", err) + } + all, err := reloaded.List(ctx, ports.BenchmarkResultFilter{}) + if err != nil { + t.Fatalf("List() after reload error = %v", err) + } + if len(all) != 1 || all[0].ID != "recent" { + t.Errorf("List() after reload = %+v, want only 'recent'", all) + } +} + +func TestNewFileBenchmarkResultStore_DefaultDir(t *testing.T) { + // Passing an empty dir should not error; it falls back to a default path + // relative to the working directory. Use t.TempDir() as cwd substitute is + // not straightforward, so just verify no panic/error using an explicit + // nested path instead. + dir := filepath.Join(t.TempDir(), "nested", "benchmarks") + store, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("NewFileBenchmarkResultStore() with nested dir error = %v", err) + } + if store == nil { + t.Fatal("NewFileBenchmarkResultStore() returned nil store") + } +} diff --git a/internal/application/services/performance/benchmark_service.go b/internal/application/services/performance/benchmark_service.go index d0a1cab..ea688f4 100644 --- a/internal/application/services/performance/benchmark_service.go +++ b/internal/application/services/performance/benchmark_service.go @@ -30,6 +30,11 @@ type BenchmarkService struct { activeRuns map[string]*BenchmarkExecution runsMutex sync.RWMutex + // Optional persistent storage for completed results. When nil, results + // only live in-memory for the RetainResults window (see activeRuns). + resultStore ports.BenchmarkResultStorePort + resultStoreMutex sync.RWMutex + // Configuration config *BenchmarkServiceConfig } @@ -119,6 +124,31 @@ func NewBenchmarkService( return service } +// SetResultStore configures persistent storage for completed benchmark +// results. It is safe to call at any time, including after the service has +// started executing benchmarks; nil disables persistence. +func (s *BenchmarkService) SetResultStore(store ports.BenchmarkResultStorePort) { + s.resultStoreMutex.Lock() + defer s.resultStoreMutex.Unlock() + s.resultStore = store +} + +func (s *BenchmarkService) getResultStore() ports.BenchmarkResultStorePort { + s.resultStoreMutex.RLock() + defer s.resultStoreMutex.RUnlock() + return s.resultStore +} + +// GetBenchmarkHistory returns persisted benchmark results matching filter. +// It returns an error if no result store has been configured. +func (s *BenchmarkService) GetBenchmarkHistory(ctx context.Context, filter ports.BenchmarkResultFilter) ([]*ports.BenchmarkResult, error) { + store := s.getResultStore() + if store == nil { + return nil, fmt.Errorf("benchmark result persistence is not configured") + } + return store.List(ctx, filter) +} + // RegisterBenchmarkTool registers a benchmark tool implementation func (s *BenchmarkService) RegisterBenchmarkTool(name string, tool ports.BenchmarkToolPort) error { s.toolsMutex.Lock() @@ -257,6 +287,14 @@ func (s *BenchmarkService) executeAsync(execution *BenchmarkExecution) { return 0 }(), }).Info("Benchmark execution completed") + + if store := s.getResultStore(); store != nil { + // Persist using a background context: the execution context may + // already be cancelled/timed out by the time we get here. + if err := store.Save(context.Background(), result); err != nil { + s.logger.WithError(err).WithField("execution_id", execution.ID).Warn("Failed to persist benchmark result") + } + } } // GetBenchmarkResult returns the result of a benchmark execution @@ -511,6 +549,14 @@ func (s *BenchmarkService) cleanupOldExecutions() { if len(toDelete) > 0 { s.logger.WithField("cleaned_count", len(toDelete)).Info("Cleaned up old benchmark executions") } + + if store := s.getResultStore(); store != nil { + if deleted, err := store.DeleteOlderThan(context.Background(), cutoff); err != nil { + s.logger.WithError(err).Warn("Failed to compact persisted benchmark results") + } else if deleted > 0 { + s.logger.WithField("deleted_count", deleted).Info("Removed persisted benchmark results past retention") + } + } } // defaultBenchmarkServiceConfig returns default configuration diff --git a/internal/application/services/performance/benchmark_service_test.go b/internal/application/services/performance/benchmark_service_test.go new file mode 100644 index 0000000..60bf31a --- /dev/null +++ b/internal/application/services/performance/benchmark_service_test.go @@ -0,0 +1,227 @@ +package performance + +import ( + "context" + "testing" + "time" + + "sql-graph-visualizer/internal/application/ports" +) + +// fakeBenchmarkTool is a minimal ports.BenchmarkToolPort implementation used +// to exercise BenchmarkService without depending on sysbench or a real +// database connection. +type fakeBenchmarkTool struct { + available bool + executeFunc func(ctx context.Context, config ports.BenchmarkConfig) (*ports.BenchmarkResult, error) + validateErr error +} + +func (f *fakeBenchmarkTool) Execute(ctx context.Context, config ports.BenchmarkConfig) (*ports.BenchmarkResult, error) { + if f.executeFunc != nil { + return f.executeFunc(ctx, config) + } + return &ports.BenchmarkResult{ + ToolName: "fake", + TestType: config.TestType, + Status: ports.BenchmarkStatusCompleted, + Metrics: &ports.PerformanceMetrics{QueriesPerSecond: 42}, + }, nil +} + +func (f *fakeBenchmarkTool) Validate(_ ports.BenchmarkConfig) error { return f.validateErr } +func (f *fakeBenchmarkTool) GetSupportedTests() []string { return []string{"fake_test"} } +func (f *fakeBenchmarkTool) IsAvailable() bool { return f.available } +func (f *fakeBenchmarkTool) GetVersion() (string, error) { return "fake/1.0", nil } + +func newTestBenchmarkService(t *testing.T) *BenchmarkService { + t.Helper() + config := defaultBenchmarkServiceConfig() + config.CleanupInterval = time.Hour // avoid the cleanup goroutine racing with test assertions + return NewBenchmarkService(nil, nil, nil, nil, newTestLogger(), config) +} + +func TestBenchmarkService_RegisterBenchmarkTool(t *testing.T) { + svc := newTestBenchmarkService(t) + + if err := svc.RegisterBenchmarkTool("unavailable", &fakeBenchmarkTool{available: false}); err == nil { + t.Error("RegisterBenchmarkTool() with unavailable tool: expected error, got nil") + } + + if err := svc.RegisterBenchmarkTool("fake", &fakeBenchmarkTool{available: true}); err != nil { + t.Fatalf("RegisterBenchmarkTool() error = %v", err) + } + + tools := svc.GetAvailableTools() + if len(tools) != 1 || tools[0] != "fake" { + t.Errorf("GetAvailableTools() = %v, want [fake]", tools) + } +} + +func TestBenchmarkService_ApplyConfigDefaults(t *testing.T) { + svc := newTestBenchmarkService(t) + svc.config.DefaultTestType = "oltp_read_write" + svc.config.DefaultDatabaseURL = "mysql://user:pass@localhost:3306/db" + svc.config.DefaultDatabaseType = "mysql" + svc.config.DefaultThreads = 4 + svc.config.DefaultTables = 2 + svc.config.DefaultTableSize = 1000 + svc.config.DefaultTestDuration = 30 * time.Second + + config := ports.BenchmarkConfig{} + svc.applyConfigDefaults(&config) + + if config.TestType != "oltp_read_write" { + t.Errorf("TestType = %q, want oltp_read_write", config.TestType) + } + if config.DatabaseURL != "mysql://user:pass@localhost:3306/db" { + t.Errorf("DatabaseURL = %q, want default", config.DatabaseURL) + } + if config.Threads != 4 || config.Tables != 2 || config.TableSize != 1000 { + t.Errorf("Threads/Tables/TableSize = %d/%d/%d, want 4/2/1000", config.Threads, config.Tables, config.TableSize) + } + if config.Duration != 30*time.Second { + t.Errorf("Duration = %v, want 30s", config.Duration) + } + + // Explicitly set fields must not be overridden. + explicit := ports.BenchmarkConfig{TestType: "oltp_read_only", Threads: 16} + svc.applyConfigDefaults(&explicit) + if explicit.TestType != "oltp_read_only" || explicit.Threads != 16 { + t.Errorf("applyConfigDefaults() overrode explicit values: %+v", explicit) + } +} + +func TestBenchmarkService_ValidateConfig(t *testing.T) { + svc := newTestBenchmarkService(t) + svc.config.MaxDuration = time.Minute + svc.config.MaxThreads = 8 + svc.config.MaxTableSize = 1000 + + tests := []struct { + name string + config ports.BenchmarkConfig + wantErr bool + }{ + {"within limits", ports.BenchmarkConfig{Duration: 30 * time.Second, Threads: 4, TableSize: 500}, false}, + {"duration exceeds max", ports.BenchmarkConfig{Duration: 2 * time.Minute, Threads: 4, TableSize: 500}, true}, + {"threads exceed max", ports.BenchmarkConfig{Duration: 30 * time.Second, Threads: 100, TableSize: 500}, true}, + {"table size exceeds max", ports.BenchmarkConfig{Duration: 30 * time.Second, Threads: 4, TableSize: 10000}, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := svc.validateConfig(tt.config) + if tt.wantErr && err == nil { + t.Error("validateConfig() expected error, got nil") + } + if !tt.wantErr && err != nil { + t.Errorf("validateConfig() unexpected error = %v", err) + } + }) + } +} + +func TestBenchmarkService_ExecuteBenchmark_PersistsResult(t *testing.T) { + svc := newTestBenchmarkService(t) + if err := svc.RegisterBenchmarkTool("fake", &fakeBenchmarkTool{available: true}); err != nil { + t.Fatalf("RegisterBenchmarkTool() error = %v", err) + } + + dir := t.TempDir() + store, err := NewFileBenchmarkResultStore(newTestLogger(), dir) + if err != nil { + t.Fatalf("NewFileBenchmarkResultStore() error = %v", err) + } + svc.SetResultStore(store) + + executionID, err := svc.ExecuteBenchmark(context.Background(), ports.BenchmarkConfig{ + TestType: "fake_test", + Duration: time.Second, + Threads: 1, + }, "fake") + if err != nil { + t.Fatalf("ExecuteBenchmark() error = %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + result, resultErr := svc.GetBenchmarkResult(executionID) + if resultErr == nil && result.Status == ports.BenchmarkStatusCompleted { + break + } + time.Sleep(10 * time.Millisecond) + } + + result, err := svc.GetBenchmarkResult(executionID) + if err != nil { + t.Fatalf("GetBenchmarkResult() error = %v", err) + } + if result.Status != ports.BenchmarkStatusCompleted { + t.Fatalf("benchmark did not complete in time, status = %v", result.Status) + } + + // The completed result should have been persisted to the store. + history, err := svc.GetBenchmarkHistory(context.Background(), ports.BenchmarkResultFilter{}) + if err != nil { + t.Fatalf("GetBenchmarkHistory() error = %v", err) + } + if len(history) != 1 || history[0].ID != executionID { + t.Errorf("GetBenchmarkHistory() = %+v, want a single entry with ID %q", history, executionID) + } +} + +func TestBenchmarkService_GetBenchmarkHistory_NoStoreConfigured(t *testing.T) { + svc := newTestBenchmarkService(t) + if _, err := svc.GetBenchmarkHistory(context.Background(), ports.BenchmarkResultFilter{}); err == nil { + t.Error("GetBenchmarkHistory() with no store configured: expected error, got nil") + } +} + +func TestBenchmarkService_ExecuteBenchmark_UnknownTool(t *testing.T) { + svc := newTestBenchmarkService(t) + if _, err := svc.ExecuteBenchmark(context.Background(), ports.BenchmarkConfig{ + TestType: "fake_test", + Duration: time.Second, + Threads: 1, + }, "does-not-exist"); err == nil { + t.Error("ExecuteBenchmark() with unregistered tool: expected error, got nil") + } +} + +func TestBenchmarkService_MaxConcurrentRuns(t *testing.T) { + svc := newTestBenchmarkService(t) + svc.config.MaxConcurrentRuns = 1 + + block := make(chan struct{}) + defer close(block) + + tool := &fakeBenchmarkTool{ + available: true, + executeFunc: func(ctx context.Context, _ ports.BenchmarkConfig) (*ports.BenchmarkResult, error) { + select { + case <-block: + case <-ctx.Done(): + } + return &ports.BenchmarkResult{Status: ports.BenchmarkStatusCompleted, Metrics: &ports.PerformanceMetrics{}}, nil + }, + } + if err := svc.RegisterBenchmarkTool("fake", tool); err != nil { + t.Fatalf("RegisterBenchmarkTool() error = %v", err) + } + + config := ports.BenchmarkConfig{TestType: "fake_test", Duration: 5 * time.Second, Threads: 1} + if _, err := svc.ExecuteBenchmark(context.Background(), config, "fake"); err != nil { + t.Fatalf("first ExecuteBenchmark() error = %v", err) + } + + // Give the async goroutine a moment to register as "running". + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) && svc.getActiveRunCount() == 0 { + time.Sleep(5 * time.Millisecond) + } + + if _, err := svc.ExecuteBenchmark(context.Background(), config, "fake"); err == nil { + t.Error("ExecuteBenchmark() exceeding MaxConcurrentRuns: expected error, got nil") + } +} diff --git a/internal/application/services/performance/custom_query_adapter.go b/internal/application/services/performance/custom_query_adapter.go new file mode 100644 index 0000000..8aa12ac --- /dev/null +++ b/internal/application/services/performance/custom_query_adapter.go @@ -0,0 +1,378 @@ +package performance + +import ( + "context" + "database/sql" + "fmt" + "math/rand" + "strings" + "sync" + "time" + + "sql-graph-visualizer/internal/application/ports" + + "github.com/sirupsen/logrus" +) + +// allowedCustomQueryPrefixes restricts custom benchmark queries to +// read/write statement types that cannot alter schema or bulk-delete data. +// This is a defense-in-depth measure: custom queries come from operator +// configuration, not end users, but running arbitrary DDL/DELETE/TRUNCATE +// statements as part of a "benchmark" is still an easy way to cause +// accidental damage. +var allowedCustomQueryPrefixes = []string{"SELECT", "INSERT", "UPDATE"} + +// CustomQueryAdapter implements ports.BenchmarkToolPort by executing +// operator-configured, named sets of SQL queries against the active source +// database via the standard database/sql connection pool (the same *sql.DB +// used elsewhere in the application, e.g. by PerformanceSchemaAdapter). +type CustomQueryAdapter struct { + logger *logrus.Logger + db *sql.DB + querySets map[string]ports.CustomBenchmarkConfig +} + +// NewCustomQueryAdapter creates a new custom query benchmark adapter. Query +// sets with no queries, or whose queries are all disallowed statement types, +// are skipped with a warning. +func NewCustomQueryAdapter(logger *logrus.Logger, db *sql.DB, querySets map[string]ports.CustomBenchmarkConfig) *CustomQueryAdapter { + return &CustomQueryAdapter{ + logger: logger, + db: db, + querySets: querySets, + } +} + +// Execute runs the custom query set named by config.CustomParams["query_set"] +// (or the sole configured set, if there is exactly one) for config.Duration +// using config.Threads concurrent workers, cycling through the set's queries +// weighted by their configured Weight. +func (c *CustomQueryAdapter) Execute(ctx context.Context, config ports.BenchmarkConfig) (*ports.BenchmarkResult, error) { + set, setName, err := c.resolveQuerySet(config) + if err != nil { + return nil, err + } + + threads := config.Threads + if threads <= 0 { + threads = set.Threads + } + if threads <= 0 { + threads = 1 + } + + duration := config.Duration + if duration <= 0 { + duration = set.Duration + } + if duration <= 0 { + duration = 30 * time.Second + } + + weights := make([]int, len(set.Queries)) + totalWeight := 0 + for i, q := range set.Queries { + w := q.Weight + if w <= 0 { + w = 1 + } + weights[i] = w + totalWeight += w + } + + stats := make([]*customQueryStat, len(set.Queries)) + for i, q := range set.Queries { + stats[i] = &customQueryStat{def: q} + } + var statsMu sync.Mutex + + startTime := time.Now() + deadline := startTime.Add(duration) + + var wg sync.WaitGroup + for w := 0; w < threads; w++ { + wg.Add(1) + go func(seed int64) { + defer wg.Done() + rnd := rand.New(rand.NewSource(seed)) //nolint:gosec // non-cryptographic query selection + for time.Now().Before(deadline) { + select { + case <-ctx.Done(): + return + default: + } + + idx := pickWeightedIndex(rnd, weights, totalWeight) + q := set.Queries[idx] + + queryStart := time.Now() + rowsExamined, rowsReturned, execErr := c.runQuery(ctx, q) + elapsed := time.Since(queryStart) + + statsMu.Lock() + stats[idx].record(elapsed, rowsExamined, rowsReturned, execErr) + statsMu.Unlock() + } + }(time.Now().UnixNano() + int64(w)) + } + wg.Wait() + + endTime := time.Now() + metrics, queryResults := aggregateCustomQueryStats(stats, endTime.Sub(startTime)) + + c.logger.WithFields(logrus.Fields{ + "query_set": setName, + "threads": threads, + "duration": duration, + "queries_per_sec": metrics.QueriesPerSecond, + "total_errors": metrics.TotalErrors, + }).Info("Custom query benchmark completed") + + return &ports.BenchmarkResult{ + ToolName: "custom", + TestType: setName, + StartTime: startTime, + EndTime: endTime, + Duration: endTime.Sub(startTime), + Metrics: metrics, + QueryResults: queryResults, + Status: ports.BenchmarkStatusCompleted, + }, nil +} + +// Validate checks that the requested query set exists, is non-empty, and +// only contains allowed statement types. +func (c *CustomQueryAdapter) Validate(config ports.BenchmarkConfig) error { + set, setName, err := c.resolveQuerySet(config) + if err != nil { + return err + } + if len(set.Queries) == 0 { + return fmt.Errorf("custom query set %q has no queries configured", setName) + } + for _, q := range set.Queries { + if !isAllowedCustomQuery(q.Query) { + return fmt.Errorf("query set %q contains a statement type that is not allowed (only SELECT/INSERT/UPDATE): %q", setName, q.Query) + } + } + if c.db == nil { + return fmt.Errorf("no database connection is configured for custom query benchmarks") + } + return nil +} + +// GetSupportedTests returns the names of configured custom query sets; each +// name can be requested via BenchmarkConfig.CustomParams["query_set"]. +func (c *CustomQueryAdapter) GetSupportedTests() []string { + names := make([]string, 0, len(c.querySets)) + for name := range c.querySets { + names = append(names, name) + } + return names +} + +// IsAvailable reports whether at least one valid query set is configured and +// a database connection is available. +func (c *CustomQueryAdapter) IsAvailable() bool { + if c.db == nil || len(c.querySets) == 0 { + return false + } + for _, set := range c.querySets { + if len(set.Queries) > 0 { + return true + } + } + return false +} + +// GetVersion returns a static identifier: this adapter has no external +// binary/version to report. +func (c *CustomQueryAdapter) GetVersion() (string, error) { + return "custom-query-adapter/1.0", nil +} + +// resolveQuerySet finds the query set requested via CustomParams["query_set"], +// falling back to the sole configured set when there is exactly one. +func (c *CustomQueryAdapter) resolveQuerySet(config ports.BenchmarkConfig) (ports.CustomBenchmarkConfig, string, error) { + name, _ := config.CustomParams["query_set"].(string) + if name == "" { + if len(c.querySets) == 1 { + for onlyName := range c.querySets { + name = onlyName + } + } else { + return ports.CustomBenchmarkConfig{}, "", fmt.Errorf("custom_params.query_set is required (available: %v)", c.GetSupportedTests()) + } + } + + set, exists := c.querySets[name] + if !exists { + return ports.CustomBenchmarkConfig{}, name, fmt.Errorf("unknown custom query set %q (available: %v)", name, c.GetSupportedTests()) + } + return set, name, nil +} + +// runQuery executes a single query definition and returns rows +// examined/returned. SELECT statements report the number of rows scanned as +// both examined and returned; INSERT/UPDATE report affected rows as +// "examined" with zero rows "returned". +func (c *CustomQueryAdapter) runQuery(ctx context.Context, q ports.CustomQueryDefinition) (rowsExamined, rowsReturned int64, err error) { + if !isAllowedCustomQuery(q.Query) { + return 0, 0, fmt.Errorf("statement type not allowed: %q", q.Query) + } + + statementType := strings.ToUpper(strings.Fields(strings.TrimSpace(q.Query))[0]) + if statementType == "SELECT" { + rows, queryErr := c.db.QueryContext(ctx, q.Query, q.Parameters...) + if queryErr != nil { + return 0, 0, queryErr + } + defer func() { _ = rows.Close() }() + + var count int64 + for rows.Next() { + count++ + } + if err := rows.Err(); err != nil { + return count, count, err + } + return count, count, nil + } + + result, execErr := c.db.ExecContext(ctx, q.Query, q.Parameters...) + if execErr != nil { + return 0, 0, execErr + } + affected, _ := result.RowsAffected() + return affected, 0, nil +} + +// isAllowedCustomQuery reports whether query begins with one of the allowed +// statement keywords. +func isAllowedCustomQuery(query string) bool { + trimmed := strings.TrimSpace(query) + if trimmed == "" { + return false + } + fields := strings.Fields(trimmed) + first := strings.ToUpper(fields[0]) + for _, allowed := range allowedCustomQueryPrefixes { + if first == allowed { + return true + } + } + return false +} + +// pickWeightedIndex selects a random index from weights (a slice of positive +// weights summing to totalWeight) using weighted random selection. Falls +// back to a uniform pick if the slice is empty or weights are non-positive. +func pickWeightedIndex(rnd *rand.Rand, weights []int, totalWeight int) int { + if len(weights) == 0 { + return 0 + } + if totalWeight <= 0 { + return rnd.Intn(len(weights)) + } + target := rnd.Intn(totalWeight) + cumulative := 0 + for i, w := range weights { + cumulative += w + if target < cumulative { + return i + } + } + return len(weights) - 1 +} + +// customQueryStat accumulates execution statistics for a single query +// definition within a benchmark run. +type customQueryStat struct { + def ports.CustomQueryDefinition + execCount int64 + errorCount int64 + totalTime time.Duration + minTime time.Duration + maxTime time.Duration + rowsExamined int64 + rowsReturned int64 +} + +func (s *customQueryStat) record(elapsed time.Duration, rowsExamined, rowsReturned int64, err error) { + s.execCount++ + s.totalTime += elapsed + if s.minTime == 0 || elapsed < s.minTime { + s.minTime = elapsed + } + if elapsed > s.maxTime { + s.maxTime = elapsed + } + if err != nil { + s.errorCount++ + return + } + s.rowsExamined += rowsExamined + s.rowsReturned += rowsReturned +} + +// aggregateCustomQueryStats converts per-query statistics into the shared +// PerformanceMetrics/QueryPerformance shapes used by all benchmark tools. +func aggregateCustomQueryStats(stats []*customQueryStat, wallClock time.Duration) (*ports.PerformanceMetrics, []ports.QueryPerformance) { + metrics := &ports.PerformanceMetrics{} + queryResults := make([]ports.QueryPerformance, 0, len(stats)) + + var totalExec int64 + var totalErrors int64 + var totalLatencyMs float64 + + for _, s := range stats { + if s.execCount == 0 { + continue + } + totalExec += s.execCount + totalErrors += s.errorCount + avgTime := s.totalTime / time.Duration(s.execCount) + totalLatencyMs += float64(avgTime.Milliseconds()) * float64(s.execCount) + + statementType := "SELECT" + if fields := strings.Fields(strings.TrimSpace(s.def.Query)); len(fields) > 0 { + statementType = strings.ToUpper(fields[0]) + } + + queryResults = append(queryResults, ports.QueryPerformance{ + QueryPattern: s.def.Query, + QueryType: statementType, + ExecutionCount: s.execCount, + TotalTime: s.totalTime, + AverageTime: avgTime, + MinTime: s.minTime, + MaxTime: s.maxTime, + RowsExamined: s.rowsExamined, + RowsReturned: s.rowsReturned, + RelationshipType: "CUSTOM_QUERY", + PerformanceImpact: classifyCustomQueryImpact(avgTime), + }) + } + + if wallClock > 0 { + metrics.QueriesPerSecond = float64(totalExec) / wallClock.Seconds() + } + if totalExec > 0 { + metrics.AverageLatency = totalLatencyMs / float64(totalExec) + metrics.ErrorRate = (float64(totalErrors) / float64(totalExec)) * 100 + } + metrics.TotalErrors = int(totalErrors) + + return metrics, queryResults +} + +func classifyCustomQueryImpact(avgLatency time.Duration) string { + switch { + case avgLatency < 10*time.Millisecond: + return "LOW" + case avgLatency < 100*time.Millisecond: + return "MEDIUM" + default: + return "HIGH" + } +} diff --git a/internal/application/services/performance/custom_query_adapter_test.go b/internal/application/services/performance/custom_query_adapter_test.go new file mode 100644 index 0000000..dd4c394 --- /dev/null +++ b/internal/application/services/performance/custom_query_adapter_test.go @@ -0,0 +1,260 @@ +package performance + +import ( + "database/sql" + "math/rand" + "testing" + "time" + + "sql-graph-visualizer/internal/application/ports" + + _ "github.com/go-sql-driver/mysql" +) + +// fakeDB returns a *sql.DB that is valid to construct (sql.Open does not +// connect eagerly) but is never actually used to run queries in these tests. +func fakeDB(t *testing.T) *sql.DB { + t.Helper() + db, err := sql.Open("mysql", "user:pass@tcp(127.0.0.1:3306)/testdb") + if err != nil { + t.Fatalf("sql.Open() error = %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + return db +} + +func TestIsAllowedCustomQuery(t *testing.T) { + tests := []struct { + query string + want bool + }{ + {"SELECT * FROM users", true}, + {" select id from users", true}, + {"INSERT INTO logs (msg) VALUES ('x')", true}, + {"update users set name = 'x' where id = 1", true}, + {"DELETE FROM users", false}, + {"DROP TABLE users", false}, + {"TRUNCATE TABLE users", false}, + {"", false}, + {" ", false}, + } + for _, tt := range tests { + if got := isAllowedCustomQuery(tt.query); got != tt.want { + t.Errorf("isAllowedCustomQuery(%q) = %v, want %v", tt.query, got, tt.want) + } + } +} + +func TestCustomQueryAdapter_ResolveQuerySet(t *testing.T) { + single := map[string]ports.CustomBenchmarkConfig{ + "only_set": {Name: "only_set", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + } + multi := map[string]ports.CustomBenchmarkConfig{ + "set_a": {Name: "set_a", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + "set_b": {Name: "set_b", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 2"}}}, + } + + t.Run("falls back to sole configured set", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), nil, single) + set, name, err := adapter.resolveQuerySet(ports.BenchmarkConfig{}) + if err != nil { + t.Fatalf("resolveQuerySet() error = %v", err) + } + if name != "only_set" || len(set.Queries) != 1 { + t.Errorf("resolveQuerySet() = (%+v, %q), want only_set", set, name) + } + }) + + t.Run("requires explicit query_set when multiple sets configured", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), nil, multi) + if _, _, err := adapter.resolveQuerySet(ports.BenchmarkConfig{}); err == nil { + t.Error("resolveQuerySet() with no query_set and multiple sets: expected error, got nil") + } + }) + + t.Run("resolves named query_set", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), nil, multi) + set, name, err := adapter.resolveQuerySet(ports.BenchmarkConfig{ + CustomParams: map[string]interface{}{"query_set": "set_b"}, + }) + if err != nil { + t.Fatalf("resolveQuerySet() error = %v", err) + } + if name != "set_b" || set.Queries[0].Query != "SELECT 2" { + t.Errorf("resolveQuerySet() = (%+v, %q), want set_b", set, name) + } + }) + + t.Run("unknown query_set errors", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), nil, multi) + if _, _, err := adapter.resolveQuerySet(ports.BenchmarkConfig{ + CustomParams: map[string]interface{}{"query_set": "does_not_exist"}, + }); err == nil { + t.Error("resolveQuerySet() with unknown name: expected error, got nil") + } + }) +} + +func TestCustomQueryAdapter_Validate(t *testing.T) { + db := fakeDB(t) + + t.Run("rejects empty query set", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), db, map[string]ports.CustomBenchmarkConfig{ + "empty": {Name: "empty"}, + }) + if err := adapter.Validate(ports.BenchmarkConfig{CustomParams: map[string]interface{}{"query_set": "empty"}}); err == nil { + t.Error("Validate() with no queries: expected error, got nil") + } + }) + + t.Run("rejects disallowed statement type", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), db, map[string]ports.CustomBenchmarkConfig{ + "dangerous": {Name: "dangerous", Queries: []ports.CustomQueryDefinition{{Query: "DELETE FROM users"}}}, + }) + if err := adapter.Validate(ports.BenchmarkConfig{CustomParams: map[string]interface{}{"query_set": "dangerous"}}); err == nil { + t.Error("Validate() with DELETE statement: expected error, got nil") + } + }) + + t.Run("rejects missing database connection", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), nil, map[string]ports.CustomBenchmarkConfig{ + "ok": {Name: "ok", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + }) + if err := adapter.Validate(ports.BenchmarkConfig{CustomParams: map[string]interface{}{"query_set": "ok"}}); err == nil { + t.Error("Validate() with nil db: expected error, got nil") + } + }) + + t.Run("accepts a well-formed query set", func(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), db, map[string]ports.CustomBenchmarkConfig{ + "ok": {Name: "ok", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + }) + if err := adapter.Validate(ports.BenchmarkConfig{CustomParams: map[string]interface{}{"query_set": "ok"}}); err != nil { + t.Errorf("Validate() unexpected error = %v", err) + } + }) +} + +func TestCustomQueryAdapter_IsAvailable(t *testing.T) { + db := fakeDB(t) + + if adapter := NewCustomQueryAdapter(newTestLogger(), nil, map[string]ports.CustomBenchmarkConfig{ + "ok": {Name: "ok", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + }); adapter.IsAvailable() { + t.Error("IsAvailable() with nil db: want false, got true") + } + + if adapter := NewCustomQueryAdapter(newTestLogger(), db, map[string]ports.CustomBenchmarkConfig{}); adapter.IsAvailable() { + t.Error("IsAvailable() with no query sets: want false, got true") + } + + adapter := NewCustomQueryAdapter(newTestLogger(), db, map[string]ports.CustomBenchmarkConfig{ + "ok": {Name: "ok", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + }) + if !adapter.IsAvailable() { + t.Error("IsAvailable() with configured db and query set: want true, got false") + } +} + +func TestPickWeightedIndex(t *testing.T) { + rnd := rand.New(rand.NewSource(1)) + + if got := pickWeightedIndex(rnd, nil, 0); got != 0 { + t.Errorf("pickWeightedIndex(empty) = %d, want 0", got) + } + + if got := pickWeightedIndex(rnd, []int{5}, 5); got != 0 { + t.Errorf("pickWeightedIndex(single) = %d, want 0", got) + } + + // With a zero/negative total weight, selection should still return a + // valid index within bounds rather than panicking. + weights := []int{1, 1, 1} + for i := 0; i < 20; i++ { + got := pickWeightedIndex(rnd, weights, 0) + if got < 0 || got >= len(weights) { + t.Fatalf("pickWeightedIndex() = %d out of bounds for weights %v", got, weights) + } + } + + // Heavily weighted first index should be picked far more often than the + // second across many trials. + heavy := []int{99, 1} + firstCount := 0 + trials := 1000 + for i := 0; i < trials; i++ { + if pickWeightedIndex(rnd, heavy, 100) == 0 { + firstCount++ + } + } + if firstCount < trials*8/10 { + t.Errorf("pickWeightedIndex() picked index 0 only %d/%d times with weight 99:1, expected it to dominate", firstCount, trials) + } +} + +func TestAggregateCustomQueryStats(t *testing.T) { + stats := []*customQueryStat{ + { + def: ports.CustomQueryDefinition{Query: "SELECT * FROM users"}, + execCount: 10, + totalTime: 100 * time.Millisecond, + minTime: 5 * time.Millisecond, + maxTime: 20 * time.Millisecond, + rowsExamined: 100, + rowsReturned: 100, + }, + { + def: ports.CustomQueryDefinition{Query: "INSERT INTO logs VALUES (1)"}, + execCount: 5, + errorCount: 1, + totalTime: 50 * time.Millisecond, + }, + { + // A query that never executed should be skipped entirely. + def: ports.CustomQueryDefinition{Query: "SELECT * FROM never_ran"}, + }, + } + + metrics, queryResults := aggregateCustomQueryStats(stats, 1*time.Second) + + if len(queryResults) != 2 { + t.Fatalf("aggregateCustomQueryStats() returned %d query results, want 2", len(queryResults)) + } + if metrics.QueriesPerSecond != 15 { + t.Errorf("QueriesPerSecond = %v, want 15", metrics.QueriesPerSecond) + } + if metrics.TotalErrors != 1 { + t.Errorf("TotalErrors = %d, want 1", metrics.TotalErrors) + } + wantErrorRate := (1.0 / 15.0) * 100 + if diff := metrics.ErrorRate - wantErrorRate; diff > 0.01 || diff < -0.01 { + t.Errorf("ErrorRate = %v, want ~%v", metrics.ErrorRate, wantErrorRate) + } +} + +func TestCustomQueryAdapter_ExecuteUnknownQuerySet(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), fakeDB(t), map[string]ports.CustomBenchmarkConfig{ + "set_a": {Name: "set_a", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + "set_b": {Name: "set_b", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 2"}}}, + }) + + if _, err := adapter.Execute(t.Context(), ports.BenchmarkConfig{}); err == nil { + t.Error("Execute() with ambiguous query_set: expected error, got nil") + } +} + +func TestCustomQueryAdapter_GetVersionAndSupportedTests(t *testing.T) { + adapter := NewCustomQueryAdapter(newTestLogger(), fakeDB(t), map[string]ports.CustomBenchmarkConfig{ + "set_a": {Name: "set_a", Queries: []ports.CustomQueryDefinition{{Query: "SELECT 1"}}}, + }) + + version, err := adapter.GetVersion() + if err != nil || version == "" { + t.Errorf("GetVersion() = (%q, %v), want non-empty version and no error", version, err) + } + + tests := adapter.GetSupportedTests() + if len(tests) != 1 || tests[0] != "set_a" { + t.Errorf("GetSupportedTests() = %v, want [set_a]", tests) + } +} diff --git a/internal/application/services/performance/graph_performance_mapper.go b/internal/application/services/performance/graph_performance_mapper.go index 7a33a5a..6270b29 100644 --- a/internal/application/services/performance/graph_performance_mapper.go +++ b/internal/application/services/performance/graph_performance_mapper.go @@ -546,6 +546,13 @@ func (gpm *GraphPerformanceMapper) identifyEdgeIssues(_ EdgePerformanceData) []E func (gpm *GraphPerformanceMapper) calculateGlobalMetrics(_ *PerformanceGraphData) {} func (gpm *GraphPerformanceMapper) identifyHotspotsAndBottlenecks(_ *PerformanceGraphData) {} +// DefaultGraphPerformanceMapperConfig returns the default graph performance +// mapper configuration. Exported so callers (e.g. bootstrap) can start from +// sane defaults and override individual fields from user configuration. +func DefaultGraphPerformanceMapperConfig() *GraphPerformanceMapperConfig { + return defaultGraphPerformanceMapperConfig() +} + // Default configuration func defaultGraphPerformanceMapperConfig() *GraphPerformanceMapperConfig { return &GraphPerformanceMapperConfig{ diff --git a/internal/application/services/performance/graph_performance_mapper_test.go b/internal/application/services/performance/graph_performance_mapper_test.go new file mode 100644 index 0000000..7c47fa0 --- /dev/null +++ b/internal/application/services/performance/graph_performance_mapper_test.go @@ -0,0 +1,65 @@ +package performance + +import ( + "context" + "testing" + + "sql-graph-visualizer/internal/domain/models" +) + +func TestGraphPerformanceMapper_MapPerformanceToGraph(t *testing.T) { + mapper := NewGraphPerformanceMapper(newTestLogger(), nil, nil, nil) + + baseGraph := &models.Graph{ + Nodes: []*models.Node{ + {Label: "User", Properties: map[string]any{"id": "1"}}, + {Label: "Team", Properties: map[string]any{"id": "2"}}, + }, + Relations: []*models.Relation{ + {Type: "MEMBER_OF", From: "1", To: "2", Properties: map[string]any{}}, + }, + } + perfData := &PerformanceSchemaData{ + ConnectionStats: &ConnectionStatistics{}, + } + + graphData, err := mapper.MapPerformanceToGraph(context.Background(), baseGraph, perfData) + if err != nil { + t.Fatalf("MapPerformanceToGraph() error = %v", err) + } + if len(graphData.Nodes) != len(baseGraph.Nodes) { + t.Errorf("MapPerformanceToGraph() returned %d nodes, want %d", len(graphData.Nodes), len(baseGraph.Nodes)) + } + if len(graphData.Edges) != len(baseGraph.Relations) { + t.Errorf("MapPerformanceToGraph() returned %d edges, want %d", len(graphData.Edges), len(baseGraph.Relations)) + } + if graphData.Metadata.NodeCount != len(baseGraph.Nodes) { + t.Errorf("Metadata.NodeCount = %d, want %d", graphData.Metadata.NodeCount, len(baseGraph.Nodes)) + } +} + +func TestGraphPerformanceMapper_MapPerformanceToGraph_RequiresInputs(t *testing.T) { + mapper := NewGraphPerformanceMapper(newTestLogger(), nil, nil, nil) + ctx := context.Background() + + if _, err := mapper.MapPerformanceToGraph(ctx, nil, &PerformanceSchemaData{}); err == nil { + t.Error("MapPerformanceToGraph() with nil base graph: expected error, got nil") + } + if _, err := mapper.MapPerformanceToGraph(ctx, &models.Graph{}, nil); err == nil { + t.Error("MapPerformanceToGraph() with nil performance data: expected error, got nil") + } +} + +func TestGraphPerformanceMapper_CreatePerformanceNode_UsesIDProperty(t *testing.T) { + mapper := NewGraphPerformanceMapper(newTestLogger(), nil, nil, nil) + + node := &models.Node{Label: "User", Properties: map[string]any{"id": "abc-123"}} + perfNode := mapper.createPerformanceNode(node, map[string]*TablePerformanceInfo{}) + + if perfNode.ID != "abc-123" { + t.Errorf("createPerformanceNode() ID = %q, want abc-123", perfNode.ID) + } + if perfNode.TableName != "User" { + t.Errorf("createPerformanceNode() TableName = %q, want User", perfNode.TableName) + } +} diff --git a/internal/application/services/performance/performance_analyzer_test.go b/internal/application/services/performance/performance_analyzer_test.go new file mode 100644 index 0000000..ac5890d --- /dev/null +++ b/internal/application/services/performance/performance_analyzer_test.go @@ -0,0 +1,123 @@ +package performance + +import ( + "context" + "testing" + + "sql-graph-visualizer/internal/application/ports" +) + +func newTestAnalyzer() *PerformanceAnalyzer { + return NewPerformanceAnalyzer(newTestLogger(), nil) +} + +func TestPerformanceAnalyzer_DetectRegressions(t *testing.T) { + analyzer := newTestAnalyzer() + ctx := context.Background() + + t.Run("requires both metrics", func(t *testing.T) { + if _, err := analyzer.DetectRegressions(ctx, nil, &ports.PerformanceMetrics{}); err == nil { + t.Error("DetectRegressions() with nil current: expected error, got nil") + } + }) + + t.Run("detects latency regression", func(t *testing.T) { + previous := &ports.PerformanceMetrics{AverageLatency: 50, QueriesPerSecond: 100} + current := &ports.PerformanceMetrics{AverageLatency: 100, QueriesPerSecond: 100} + + regressions, err := analyzer.DetectRegressions(ctx, current, previous) + if err != nil { + t.Fatalf("DetectRegressions() error = %v", err) + } + found := false + for _, r := range regressions { + if r.MetricName == "average_latency" { + found = true + if r.RegressionAmount <= 0 { + t.Errorf("RegressionAmount = %v, want positive", r.RegressionAmount) + } + } + } + if !found { + t.Errorf("DetectRegressions() = %+v, want a latency regression to be detected", regressions) + } + }) + + t.Run("no regression when performance improves", func(t *testing.T) { + previous := &ports.PerformanceMetrics{AverageLatency: 100, QueriesPerSecond: 50} + current := &ports.PerformanceMetrics{AverageLatency: 50, QueriesPerSecond: 100} + + regressions, err := analyzer.DetectRegressions(ctx, current, previous) + if err != nil { + t.Fatalf("DetectRegressions() error = %v", err) + } + if len(regressions) != 0 { + t.Errorf("DetectRegressions() = %+v, want no regressions when performance improves", regressions) + } + }) +} + +func TestPerformanceAnalyzer_CalculatePerformanceScore(t *testing.T) { + analyzer := newTestAnalyzer() + ctx := context.Background() + + if _, err := analyzer.CalculatePerformanceScore(ctx, nil); err == nil { + t.Error("CalculatePerformanceScore(nil) expected error, got nil") + } + + good, err := analyzer.CalculatePerformanceScore(ctx, &ports.PerformanceMetrics{ + AverageLatency: 5, QueriesPerSecond: 2000, ErrorRate: 0, + }) + if err != nil { + t.Fatalf("CalculatePerformanceScore() error = %v", err) + } + poor, err := analyzer.CalculatePerformanceScore(ctx, &ports.PerformanceMetrics{ + AverageLatency: 500, QueriesPerSecond: 1, ErrorRate: 10, + }) + if err != nil { + t.Fatalf("CalculatePerformanceScore() error = %v", err) + } + + if good.OverallScore <= poor.OverallScore { + t.Errorf("expected good metrics score (%v) > poor metrics score (%v)", good.OverallScore, poor.OverallScore) + } +} + +func TestPerformanceAnalyzer_IdentifyBottlenecks(t *testing.T) { + analyzer := newTestAnalyzer() + ctx := context.Background() + + if _, err := analyzer.IdentifyBottlenecks(ctx, nil); err == nil { + t.Error("IdentifyBottlenecks(nil) expected error, got nil") + } + + result := &ports.BenchmarkResult{ + Metrics: &ports.PerformanceMetrics{AverageLatency: 5000, QueriesPerSecond: 1}, + } + bottlenecks, err := analyzer.IdentifyBottlenecks(ctx, result) + if err != nil { + t.Fatalf("IdentifyBottlenecks() error = %v", err) + } + if len(bottlenecks) == 0 { + t.Error("IdentifyBottlenecks() with clearly bad metrics: expected at least one bottleneck") + } +} + +func TestPerformanceAnalyzer_ClassifyRegressionSeverity(t *testing.T) { + analyzer := newTestAnalyzer() + + tests := []struct { + pct float64 + want ports.SeverityLevel + }{ + {60, ports.SeverityCritical}, + {30, ports.SeverityHigh}, + {15, ports.SeverityMedium}, + {5, ports.SeverityLow}, + } + for _, tt := range tests { + if got := analyzer.classifyRegressionSeverity(tt.pct); got != tt.want { + t.Errorf("classifyRegressionSeverity(%v) = %v, want %v", tt.pct, got, tt.want) + } + } +} diff --git a/internal/application/services/performance/realtime_performance_monitor.go b/internal/application/services/performance/realtime_performance_monitor.go index a1523eb..857a9f1 100644 --- a/internal/application/services/performance/realtime_performance_monitor.go +++ b/internal/application/services/performance/realtime_performance_monitor.go @@ -628,6 +628,13 @@ func (rpm *RealtimePerformanceMonitor) GetLastGraphData() *PerformanceGraphData return rpm.lastGraphData } +// DefaultRealtimeMonitorConfig returns the default real-time monitor +// configuration. Exported so callers (e.g. bootstrap) can start from sane +// defaults and override individual fields from user configuration. +func DefaultRealtimeMonitorConfig() *RealtimeMonitorConfig { + return defaultRealtimeMonitorConfig() +} + // Default configuration func defaultRealtimeMonitorConfig() *RealtimeMonitorConfig { return &RealtimeMonitorConfig{ diff --git a/internal/application/services/performance/sysbench_adapter_test.go b/internal/application/services/performance/sysbench_adapter_test.go new file mode 100644 index 0000000..3998aa9 --- /dev/null +++ b/internal/application/services/performance/sysbench_adapter_test.go @@ -0,0 +1,173 @@ +package performance + +import ( + "testing" + "time" + + "sql-graph-visualizer/internal/application/ports" +) + +func TestSysbenchAdapter_ExtractFloat(t *testing.T) { + adapter := &SysbenchAdapter{logger: newTestLogger()} + + tests := []struct { + text string + pattern string + want float64 + }{ + {"queries: 1234.56 queries/sec", `([0-9]+\.?[0-9]*)\s*queries/sec`, 1234.56}, + {"avg: 12.34", `avg:\s*([0-9]+\.?[0-9]*)`, 12.34}, + {"no match here", `avg:\s*([0-9]+\.?[0-9]*)`, 0}, + } + + for _, tt := range tests { + if got := adapter.extractFloat(tt.text, tt.pattern); got != tt.want { + t.Errorf("extractFloat(%q, %q) = %v, want %v", tt.text, tt.pattern, got, tt.want) + } + } +} + +func TestSysbenchAdapter_ParseOutput(t *testing.T) { + adapter := &SysbenchAdapter{logger: newTestLogger()} + output := ` +SQL statistics: + queries performed: + read: 140000 + write: 40000 + total: 180000 + transactions: 18000 (300.00 transactions/sec) + queries: 180000 (3000.00 queries/sec) + ignored errors: 0 (0.00 errors/s) + +Latency (ms): + min: 1.20 + avg: 5.43 + max: 87.10 + 95th percentile: 12.50 + 99th percentile: 45.00 +` + config := ports.BenchmarkConfig{TestType: "oltp_read_write"} + metrics, queryResults, err := adapter.parseOutput(output, config) + if err != nil { + t.Fatalf("parseOutput() error = %v", err) + } + + if metrics.QueriesPerSecond != 3000.00 { + t.Errorf("QueriesPerSecond = %v, want 3000.00", metrics.QueriesPerSecond) + } + if metrics.TransactionsPerSec != 300.00 { + t.Errorf("TransactionsPerSec = %v, want 300.00", metrics.TransactionsPerSec) + } + if metrics.AverageLatency != 5.43 { + t.Errorf("AverageLatency = %v, want 5.43", metrics.AverageLatency) + } + if metrics.MinLatency != 1.20 { + t.Errorf("MinLatency = %v, want 1.20", metrics.MinLatency) + } + if metrics.MaxLatency != 87.10 { + t.Errorf("MaxLatency = %v, want 87.10", metrics.MaxLatency) + } + if metrics.Percentile95 != 12.50 { + t.Errorf("Percentile95 = %v, want 12.50", metrics.Percentile95) + } + if metrics.Percentile99 != 45.00 { + t.Errorf("Percentile99 = %v, want 45.00", metrics.Percentile99) + } + + if len(queryResults) == 0 { + t.Error("parseOutput() returned no query results for oltp_read_write") + } +} + +func TestSysbenchAdapter_GetSupportedTests(t *testing.T) { + adapter := &SysbenchAdapter{logger: newTestLogger()} + tests := adapter.GetSupportedTests() + if len(tests) == 0 { + t.Fatal("GetSupportedTests() returned no tests") + } + found := false + for _, tt := range tests { + if tt == "oltp_read_write" { + found = true + } + } + if !found { + t.Errorf("GetSupportedTests() = %v, want it to include oltp_read_write", tests) + } +} + +func TestSysbenchAdapter_Validate(t *testing.T) { + adapter := &SysbenchAdapter{logger: newTestLogger()} + + tests := []struct { + name string + config ports.BenchmarkConfig + wantErr bool + }{ + { + name: "valid oltp config", + config: ports.BenchmarkConfig{ + TestType: "oltp_read_write", DatabaseURL: "mysql://u:p@h:3306/db", + Threads: 4, Duration: 30 * time.Second, TableSize: 1000, Tables: 2, + }, + wantErr: false, + }, + {"unsupported test type", ports.BenchmarkConfig{TestType: "not_a_real_test", DatabaseURL: "x", Threads: 1, Duration: time.Second}, true}, + {"missing database url", ports.BenchmarkConfig{TestType: "oltp_read_only", Threads: 1, Duration: time.Second}, true}, + {"non-positive threads", ports.BenchmarkConfig{TestType: "oltp_read_only", DatabaseURL: "x", Threads: 0, Duration: time.Second}, true}, + {"non-positive duration", ports.BenchmarkConfig{TestType: "oltp_read_only", DatabaseURL: "x", Threads: 1, Duration: 0}, true}, + { + name: "oltp test missing table size", + config: ports.BenchmarkConfig{TestType: "oltp_read_write", DatabaseURL: "x", Threads: 1, Duration: time.Second, Tables: 2}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := adapter.Validate(tt.config) + if tt.wantErr && err == nil { + t.Error("Validate() expected error, got nil") + } + if !tt.wantErr && err != nil { + t.Errorf("Validate() unexpected error = %v", err) + } + }) + } +} + +func TestSysbenchAdapter_ParseDatabaseURL(t *testing.T) { + adapter := &SysbenchAdapter{logger: newTestLogger()} + + args, err := adapter.parseDatabaseURL("mysql://user:pass@localhost:3306/mydb", "mysql") + if err != nil { + t.Fatalf("parseDatabaseURL() error = %v", err) + } + + want := map[string]bool{ + "--db-user=user": true, + "--db-password=pass": true, + "--db-host=localhost": true, + "--db-port=3306": true, + "--db-name=mydb": true, + "--db-driver=mysql": true, + } + if len(args) != len(want) { + t.Errorf("parseDatabaseURL() returned %d args, want %d: %v", len(args), len(want), args) + } + for _, arg := range args { + if !want[arg] { + t.Errorf("parseDatabaseURL() produced unexpected arg %q", arg) + } + } +} + +func TestSysbenchAdapter_IsAvailableAndVersion_WhenBinaryMissing(t *testing.T) { + adapter := NewSysbenchAdapter(newTestLogger(), &SysbenchConfig{BinaryPath: "/nonexistent/sysbench-binary"}) + if adapter.IsAvailable() { + t.Error("IsAvailable() with nonexistent binary: want false, got true") + } + if _, err := adapter.GetVersion(); err == nil { + t.Error("GetVersion() with unavailable adapter: expected error, got nil") + } +} diff --git a/internal/domain/models/config.go b/internal/domain/models/config.go index dc25aa5..a7f59f4 100644 --- a/internal/domain/models/config.go +++ b/internal/domain/models/config.go @@ -276,12 +276,41 @@ type AlertConfig struct { // BenchmarksConfig contains benchmarking settings type BenchmarksConfig struct { - Enabled bool `yaml:"enabled"` - DefaultDuration string `yaml:"default_duration"` - MaxDuration string `yaml:"max_duration"` - ResultsRetention string `yaml:"results_retention"` - Sysbench *SysbenchConfig `yaml:"sysbench,omitempty"` - Limits *LimitsConfig `yaml:"limits,omitempty"` + Enabled bool `yaml:"enabled"` + DefaultDuration string `yaml:"default_duration"` + MaxDuration string `yaml:"max_duration"` + ResultsRetention string `yaml:"results_retention"` + // ResultsDir is the directory where benchmark results are persisted + // (JSONL file storage). Defaults to "data/performance/benchmarks" when + // unset. + ResultsDir string `yaml:"results_dir,omitempty"` + Sysbench *SysbenchConfig `yaml:"sysbench,omitempty"` + Limits *LimitsConfig `yaml:"limits,omitempty"` + // CustomQueries defines named sets of user-provided queries that can be + // benchmarked via the "custom" tool, in addition to sysbench. + CustomQueries []CustomQueryBenchmarkConfig `yaml:"custom_queries,omitempty"` +} + +// CustomQueryBenchmarkConfig defines a named, user-configurable set of +// queries to run against the active source database for benchmarking. +type CustomQueryBenchmarkConfig struct { + Name string `yaml:"name"` + Description string `yaml:"description,omitempty"` + Duration string `yaml:"duration,omitempty"` + Threads int `yaml:"threads,omitempty"` + Queries []CustomQueryDefinitionConfig `yaml:"queries"` +} + +// CustomQueryDefinitionConfig defines a single query within a custom query +// benchmark set. Only SELECT/INSERT/UPDATE statements are permitted at +// execution time regardless of what is configured here. +type CustomQueryDefinitionConfig struct { + Query string `yaml:"query"` + Weight int `yaml:"weight,omitempty"` + Description string `yaml:"description,omitempty"` + Parameters []interface{} `yaml:"parameters,omitempty"` + ExpectedLatency string `yaml:"expected_latency,omitempty"` + TargetQPS float64 `yaml:"target_qps,omitempty"` } // SysbenchConfig contains Sysbench-specific settings diff --git a/internal/interfaces/api/performance_handlers.go b/internal/interfaces/api/performance_handlers.go index 74b3f63..f14be9e 100644 --- a/internal/interfaces/api/performance_handlers.go +++ b/internal/interfaces/api/performance_handlers.go @@ -2,14 +2,17 @@ package api //nolint:revive // api is a clear and conventional package name import ( + "encoding/csv" "encoding/json" "fmt" "net/http" "strconv" + "strings" "time" "sql-graph-visualizer/internal/application/ports" "sql-graph-visualizer/internal/application/services/performance" + "sql-graph-visualizer/internal/domain/aggregates/graph" "sql-graph-visualizer/internal/domain/models" "github.com/gorilla/mux" @@ -24,6 +27,7 @@ type PerformanceHandlers struct { graphMapper *performance.GraphPerformanceMapper realtimeMonitor *performance.RealtimePerformanceMonitor psAdapter *performance.PerformanceSchemaAdapter + neo4jRepo ports.Neo4jPort } // Response represents an API response structure. @@ -56,6 +60,8 @@ type BenchmarkRequest struct { WarmupSeconds int `json:"warmup_seconds,omitempty"` DatabaseURL string `json:"database_url,omitempty"` DatabaseType string `json:"database_type,omitempty"` + // QuerySet selects a named custom query set when Tool is "custom". + QuerySet string `json:"query_set,omitempty"` Config map[string]interface{} `json:"config"` Duration int `json:"duration_seconds"` @@ -107,6 +113,7 @@ func NewPerformanceHandlers( graphMapper *performance.GraphPerformanceMapper, realtimeMonitor *performance.RealtimePerformanceMonitor, psAdapter *performance.PerformanceSchemaAdapter, + neo4jRepo ports.Neo4jPort, ) *PerformanceHandlers { return &PerformanceHandlers{ logger: logger, @@ -115,9 +122,48 @@ func NewPerformanceHandlers( graphMapper: graphMapper, realtimeMonitor: realtimeMonitor, psAdapter: psAdapter, + neo4jRepo: neo4jRepo, } } +// fetchBaseGraph loads the current domain graph from Neo4j (the same data +// source used by the main visualization's /api/graph endpoint) and converts +// it into the simplified models.Graph shape expected by GraphPerformanceMapper. +func (ph *PerformanceHandlers) fetchBaseGraph() (*models.Graph, error) { + if ph.neo4jRepo == nil { + return nil, fmt.Errorf("neo4j repository is not configured") + } + + graphInterface, err := ph.neo4jRepo.ExportGraph("MATCH (n)-[r]->(m) RETURN n, r, m") + if err != nil { + return nil, fmt.Errorf("failed to export graph from Neo4j: %w", err) + } + g, ok := graphInterface.(*graph.GraphAggregate) + if !ok { + return nil, fmt.Errorf("unexpected graph type returned from Neo4j export") + } + + baseGraph := &models.Graph{ + Nodes: make([]*models.Node, 0, len(g.GetNodes())), + Relations: make([]*models.Relation, 0, len(g.GetRelationships())), + } + for _, node := range g.GetNodes() { + baseGraph.Nodes = append(baseGraph.Nodes, &models.Node{ + Label: node.Type, + Properties: node.Properties, + }) + } + for _, rel := range g.GetRelationships() { + baseGraph.Relations = append(baseGraph.Relations, &models.Relation{ + Type: rel.Type, + From: fmt.Sprintf("%v", rel.SourceNode.ID), + To: fmt.Sprintf("%v", rel.TargetNode.ID), + Properties: rel.Properties, + }) + } + return baseGraph, nil +} + // RegisterRoutes registers all performance-related routes func (ph *PerformanceHandlers) RegisterRoutes(router *mux.Router) { // Benchmark control endpoints @@ -147,6 +193,10 @@ func (ph *PerformanceHandlers) RegisterRoutes(router *mux.Router) { // Configuration endpoints router.HandleFunc("/api/performance/config", ph.GetPerformanceConfig).Methods("GET") router.HandleFunc("/api/performance/config", ph.UpdatePerformanceConfig).Methods("PUT") + + // Reporting and export endpoints + router.HandleFunc("/api/performance/reports/summary", ph.GetPerformanceReport).Methods("GET") + router.HandleFunc("/api/performance/export", ph.ExportPerformanceData).Methods("GET") } // Benchmark control handlers @@ -176,6 +226,14 @@ func (ph *PerformanceHandlers) StartBenchmark(w http.ResponseWriter, r *http.Req // configured default. tool, testType := resolveBenchmarkRequest(req) + customParams := req.Config + if req.QuerySet != "" { + if customParams == nil { + customParams = make(map[string]interface{}) + } + customParams["query_set"] = req.QuerySet + } + config := ports.BenchmarkConfig{ TestType: testType, Duration: time.Duration(req.Duration) * time.Second, @@ -185,7 +243,7 @@ func (ph *PerformanceHandlers) StartBenchmark(w http.ResponseWriter, r *http.Req WarmupTime: time.Duration(req.WarmupSeconds) * time.Second, DatabaseType: req.DatabaseType, DatabaseURL: req.DatabaseURL, - CustomParams: req.Config, + CustomParams: customParams, } executionID, err := ph.benchmarkService.ExecuteBenchmark(r.Context(), config, tool) @@ -360,11 +418,14 @@ func (ph *PerformanceHandlers) GetCurrentPerformanceData(w http.ResponseWriter, // Include graph data if requested if includeGraph { - var baseGraph *models.Graph - if baseGraph != nil { - graphData, err := ph.graphMapper.MapPerformanceToGraph(r.Context(), baseGraph, perfData) - if err == nil { + if baseGraph, graphErr := ph.fetchBaseGraph(); graphErr != nil { + ph.logger.WithError(graphErr).Warn("Failed to load base graph for performance data response") + } else { + graphData, mapErr := ph.graphMapper.MapPerformanceToGraph(r.Context(), baseGraph, perfData) + if mapErr == nil { response.GraphData = graphData + } else { + ph.logger.WithError(mapErr).Warn("Failed to map performance data to graph") } } } @@ -421,20 +482,37 @@ func (ph *PerformanceHandlers) GetPerformanceHistory(w http.ResponseWriter, r *h } } - historyData := []interface{}{ - map[string]interface{}{ - "message": "Historical data collection not yet implemented", - "params": map[string]interface{}{ - "start_time": startTime, - "end_time": endTime, - "limit": limit, + // Historical performance data currently comes from persisted benchmark + // results (see BenchmarkResultStorePort); live Performance Schema + // snapshots are not persisted. Optional tool/test_type filters narrow the + // result set. + filter := ports.BenchmarkResultFilter{ + ToolName: r.URL.Query().Get("tool"), + TestType: r.URL.Query().Get("test_type"), + Since: startTime, + Until: endTime, + Limit: limit, + } + + results, err := ph.benchmarkService.GetBenchmarkHistory(r.Context(), filter) + if err != nil { + ph.sendJSONResponse(w, http.StatusOK, Response{ + Success: true, + Data: map[string]interface{}{ + "results": []interface{}{}, + "message": err.Error(), }, - }, + Timestamp: time.Now(), + }) + return } ph.sendJSONResponse(w, http.StatusOK, Response{ - Success: true, - Data: historyData, + Success: true, + Data: map[string]interface{}{ + "results": results, + "count": len(results), + }, Timestamp: time.Now(), }) } @@ -470,9 +548,9 @@ func (ph *PerformanceHandlers) GetPerformanceGraph(w http.ResponseWriter, r *htt return } - var baseGraph *models.Graph - if baseGraph == nil { - ph.sendErrorResponse(w, http.StatusServiceUnavailable, "graph_unavailable", "Base graph is not available", "") + baseGraph, err := ph.fetchBaseGraph() + if err != nil { + ph.sendErrorResponse(w, http.StatusServiceUnavailable, "graph_unavailable", "Base graph is not available", err.Error()) return } @@ -489,6 +567,227 @@ func (ph *PerformanceHandlers) GetPerformanceGraph(w http.ResponseWriter, r *htt }) } +// PerformanceReport is the response body for GET /api/performance/reports/summary. +type PerformanceReport struct { + GeneratedAt time.Time `json:"generated_at"` + TimeRange PerformanceReportRange `json:"time_range"` + RunsAnalyzed int `json:"runs_analyzed"` + LatestRun *ports.BenchmarkResult `json:"latest_run,omitempty"` + OverallScore *ports.PerformanceScore `json:"overall_score,omitempty"` + Bottlenecks []ports.PerformanceBottleneck `json:"bottlenecks"` + Hotspots []ports.HotspotNode `json:"hotspots"` + QueryPatterns *ports.QueryPatternAnalysis `json:"query_patterns,omitempty"` + Issues []ports.PerformanceIssue `json:"issues"` + Regressions []ports.PerformanceRegression `json:"regressions"` + OptimizationTips []ports.OptimizationSuggestion `json:"optimization_suggestions"` + Message string `json:"message,omitempty"` +} + +// PerformanceReportRange describes the time window a report covers. +type PerformanceReportRange struct { + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` +} + +// GetPerformanceReport handles requests for a summarized performance report +// (executive summary, bottlenecks, hotspots, query patterns, optimization +// suggestions, and regression detection) built from persisted benchmark +// history. Requires benchmark result persistence to be configured. +func (ph *PerformanceHandlers) GetPerformanceReport(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + + limit := 20 + if limitStr := r.URL.Query().Get("limit"); limitStr != "" { + if parsed, parseErr := strconv.Atoi(limitStr); parseErr == nil && parsed > 0 { + limit = parsed + } + } + + filter := ports.BenchmarkResultFilter{ + ToolName: r.URL.Query().Get("tool"), + TestType: r.URL.Query().Get("test_type"), + Limit: limit, + } + + history, err := ph.benchmarkService.GetBenchmarkHistory(ctx, filter) + if err != nil { + ph.sendJSONResponse(w, http.StatusOK, Response{ + Success: true, + Data: PerformanceReport{ + GeneratedAt: time.Now(), + Message: err.Error(), + }, + Timestamp: time.Now(), + }) + return + } + if len(history) == 0 { + ph.sendJSONResponse(w, http.StatusOK, Response{ + Success: true, + Data: PerformanceReport{ + GeneratedAt: time.Now(), + Message: "no benchmark runs recorded yet", + }, + Timestamp: time.Now(), + }) + return + } + + latest := history[len(history)-1] + report := PerformanceReport{ + GeneratedAt: time.Now(), + TimeRange: PerformanceReportRange{ + StartTime: history[0].StartTime, + EndTime: latest.EndTime, + }, + RunsAnalyzed: len(history), + LatestRun: latest, + } + + if bottlenecks, bErr := ph.performanceAnalyzer.IdentifyBottlenecks(ctx, latest); bErr == nil { + report.Bottlenecks = bottlenecks + } else { + ph.logger.WithError(bErr).Warn("Failed to identify bottlenecks for performance report") + } + + if queryPatterns, qErr := ph.performanceAnalyzer.AnalyzeQueryPatterns(ctx, latest.QueryResults); qErr == nil { + report.QueryPatterns = queryPatterns + } else { + ph.logger.WithError(qErr).Warn("Failed to analyze query patterns for performance report") + } + + if issues, iErr := ph.performanceAnalyzer.IdentifyInefficiencies(ctx, latest.QueryResults); iErr == nil { + report.Issues = issues + } + + if latest.Metrics != nil { + if score, sErr := ph.performanceAnalyzer.CalculatePerformanceScore(ctx, latest.Metrics); sErr == nil { + report.OverallScore = score + } + } + + metricsHistory := make([]*ports.PerformanceMetrics, 0, len(history)) + for _, run := range history { + if run.Metrics != nil { + metricsHistory = append(metricsHistory, run.Metrics) + } + } + if hotspots, hErr := ph.performanceAnalyzer.DetectHotspots(ctx, metricsHistory); hErr == nil { + report.Hotspots = hotspots + } + + if len(history) >= 2 && latest.Metrics != nil { + previous := history[len(history)-2] + if previous.Metrics != nil { + if regressions, rErr := ph.performanceAnalyzer.DetectRegressions(ctx, latest.Metrics, previous.Metrics); rErr == nil { + report.Regressions = regressions + } + } + } + + analysis := &ports.PerformanceAnalysis{ + OverallScore: report.OverallScore, + Bottlenecks: report.Bottlenecks, + Hotspots: report.Hotspots, + QueryPatterns: report.QueryPatterns, + Issues: report.Issues, + AnalyzedAt: time.Now(), + } + if suggestions, oErr := ph.performanceAnalyzer.GenerateOptimizationSuggestions(ctx, analysis); oErr == nil { + report.OptimizationTips = suggestions + } + + ph.sendJSONResponse(w, http.StatusOK, Response{ + Success: true, + Data: report, + Timestamp: time.Now(), + }) +} + +// ExportPerformanceData handles requests to export persisted benchmark +// history as JSON or CSV, via ?format=json|csv (default json). +func (ph *PerformanceHandlers) ExportPerformanceData(w http.ResponseWriter, r *http.Request) { + format := strings.ToLower(r.URL.Query().Get("format")) + if format == "" { + format = "json" + } + + limit := 1000 + if limitStr := r.URL.Query().Get("limit"); limitStr != "" { + if parsed, parseErr := strconv.Atoi(limitStr); parseErr == nil && parsed > 0 { + limit = parsed + } + } + + filter := ports.BenchmarkResultFilter{ + ToolName: r.URL.Query().Get("tool"), + TestType: r.URL.Query().Get("test_type"), + Limit: limit, + } + + results, err := ph.benchmarkService.GetBenchmarkHistory(r.Context(), filter) + if err != nil { + ph.sendErrorResponse(w, http.StatusServiceUnavailable, "persistence_unavailable", "Benchmark result persistence is not configured", err.Error()) + return + } + + switch format { + case "csv": + ph.exportPerformanceCSV(w, results) + case "json": + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Disposition", `attachment; filename="benchmark_results.json"`) + if encErr := json.NewEncoder(w).Encode(results); encErr != nil { + ph.logger.WithError(encErr).Error("Failed to encode performance export as JSON") + } + default: + ph.sendErrorResponse(w, http.StatusBadRequest, "invalid_format", "Unsupported export format", fmt.Sprintf("format %q is not supported; use json or csv", format)) + } +} + +func (ph *PerformanceHandlers) exportPerformanceCSV(w http.ResponseWriter, results []*ports.BenchmarkResult) { + w.Header().Set("Content-Type", "text/csv") + w.Header().Set("Content-Disposition", `attachment; filename="benchmark_results.csv"`) + + csvWriter := csv.NewWriter(w) + defer csvWriter.Flush() + + header := []string{ + "id", "tool_name", "test_type", "status", "start_time", "end_time", "duration_seconds", + "queries_per_second", "average_latency_ms", "error_rate_percent", "total_errors", + } + if err := csvWriter.Write(header); err != nil { + ph.logger.WithError(err).Error("Failed to write CSV header for performance export") + return + } + + for _, result := range results { + row := []string{ + result.ID, + result.ToolName, + result.TestType, + string(result.Status), + result.StartTime.Format(time.RFC3339), + result.EndTime.Format(time.RFC3339), + strconv.FormatFloat(result.Duration.Seconds(), 'f', 3, 64), + } + if result.Metrics != nil { + row = append(row, + strconv.FormatFloat(result.Metrics.QueriesPerSecond, 'f', 3, 64), + strconv.FormatFloat(result.Metrics.AverageLatency, 'f', 3, 64), + strconv.FormatFloat(result.Metrics.ErrorRate, 'f', 3, 64), + strconv.Itoa(result.Metrics.TotalErrors), + ) + } else { + row = append(row, "", "", "", "") + } + if err := csvWriter.Write(row); err != nil { + ph.logger.WithError(err).Error("Failed to write CSV row for performance export") + return + } + } +} + // Real-time .monitoring handlers // GetRealtimeClients returns information about realtime clients. diff --git a/internal/interfaces/api/performance_handlers_test.go b/internal/interfaces/api/performance_handlers_test.go new file mode 100644 index 0000000..283a115 --- /dev/null +++ b/internal/interfaces/api/performance_handlers_test.go @@ -0,0 +1,57 @@ +package api + +import "testing" + +func TestResolveBenchmarkRequest(t *testing.T) { + tests := []struct { + name string + req BenchmarkRequest + wantTool string + wantTestType string + }{ + { + name: "explicit tool wins over benchmark_type", + req: BenchmarkRequest{Tool: "custom", BenchmarkType: "sysbench"}, + wantTool: "custom", + }, + { + name: "legacy benchmark_type carrying a tool name", + req: BenchmarkRequest{BenchmarkType: "sysbench"}, + wantTool: "sysbench", + }, + { + name: "legacy benchmark_type carrying a tool name (custom)", + req: BenchmarkRequest{BenchmarkType: "custom"}, + wantTool: "custom", + }, + { + name: "legacy benchmark_type carrying a sysbench test type", + req: BenchmarkRequest{BenchmarkType: "oltp_read_write"}, + wantTool: "sysbench", + wantTestType: "oltp_read_write", + }, + { + name: "explicit test type is preserved alongside legacy benchmark_type", + req: BenchmarkRequest{BenchmarkType: "oltp_read_write", TestType: "oltp_point_select"}, + wantTool: "sysbench", + wantTestType: "oltp_point_select", + }, + { + name: "no fields set defaults to sysbench", + req: BenchmarkRequest{}, + wantTool: "sysbench", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tool, testType := resolveBenchmarkRequest(tt.req) + if tool != tt.wantTool { + t.Errorf("resolveBenchmarkRequest() tool = %q, want %q", tool, tt.wantTool) + } + if testType != tt.wantTestType { + t.Errorf("resolveBenchmarkRequest() testType = %q, want %q", testType, tt.wantTestType) + } + }) + } +} diff --git a/internal/interfaces/web/static/js/visualization.js b/internal/interfaces/web/static/js/visualization.js index d87abe4..730f273 100644 --- a/internal/interfaces/web/static/js/visualization.js +++ b/internal/interfaces/web/static/js/visualization.js @@ -9,6 +9,12 @@ class GraphVisualizer { constructor() { this.viz = null; this.network = null; + this.currentGraphNodes = []; + this.currentGraphEdges = []; + this.originalNodeVisuals = new Map(); + this.originalEdgeVisuals = new Map(); + this.performanceOverlayActive = false; + this.performanceData = null; this.initialize(); } @@ -32,6 +38,11 @@ class GraphVisualizer { const nodes = new vis.DataSet(); const edges = new vis.DataSet(); + this.currentGraphNodes = []; + this.currentGraphEdges = []; + this.originalNodeVisuals = new Map(); + this.originalEdgeVisuals = new Map(); + if (graphData.nodes) { graphData.nodes.forEach(node => { const displayLabel = node.properties.name || @@ -71,14 +82,21 @@ class GraphVisualizer { size: nodeSize, properties: node.properties }); + + this.currentGraphNodes.push({ + id: node.id, + label: node.label, + properties: node.properties || {} + }); }); } if (graphData.relationships) { - graphData.relationships.forEach(rel => { + graphData.relationships.forEach((rel, relIndex) => { let edgeColor = '#848484'; let edgeWidth = 2; let edgeLabel = rel.type; + const edgeId = `edge-${relIndex}`; switch(rel.type) { case 'LEADS': edgeColor = '#D0021B'; edgeWidth = 3; break; @@ -102,6 +120,7 @@ class GraphVisualizer { } edges.add({ + id: edgeId, from: rel.from, to: rel.to, label: edgeLabel, @@ -113,9 +132,18 @@ class GraphVisualizer { width: edgeWidth, arrows: { to: { enabled: true, scaleFactor: 0.8 } } }); + + this.currentGraphEdges.push({ + id: edgeId, + from: rel.from, + to: rel.to, + type: rel.type + }); }); } + this.captureOriginalVisuals(nodes, edges); + console.log('Processed nodes:', nodes.get().length); console.log('Processed edges:', edges.get().length); @@ -137,12 +165,33 @@ class GraphVisualizer { this.applyThemeToNetwork(); + const perfToggle = document.getElementById('perfOverlayToggle'); + if (perfToggle && perfToggle.checked) { + this.enablePerformanceOverlay(); + } } catch (error) { console.error('Error initializing visualization:', error); } } + captureOriginalVisuals(nodes, edges) { + nodes.get().forEach(n => { + this.originalNodeVisuals.set(n.id, { + size: n.size, + color: n.color, + borderWidth: n.borderWidth, + shadow: n.shadow + }); + }); + edges.get().forEach(e => { + this.originalEdgeVisuals.set(e.id, { + width: e.width, + color: e.color + }); + }); + } + getThemeOptions() { const isDark = document.body.getAttribute("data-theme") === "dark"; @@ -189,6 +238,7 @@ class GraphVisualizer { this.initializeControlButtons(); this.initializeSearch(); this.initializeLayoutSelector(); + this.initializePerformanceOverlay(); } //////////////// @@ -497,6 +547,360 @@ class GraphVisualizer { const container = this.network.body.container; container.style.backgroundColor = isDark ? '#121212' : '#ffffff'; } + + //////////////// + // Performance overlay + // + // The performance overlay fetches /api/performance/data/graph and maps the + // returned performance metrics onto the existing vis-network nodes/edges + // (size/color/border for nodes, thickness/color for edges), without + // rebuilding the graph. Matching between the domain graph (/api/graph, + // Neo4j-based ids) and the performance graph (table/label-derived ids) is + // best-effort; any node/edge that cannot be confidently matched is simply + // left with its original appearance (graceful no-op, no thrown errors). + + initializePerformanceOverlay() { + const toggle = document.getElementById('perfOverlayToggle'); + const select = document.getElementById('perfMetricSelect'); + if (!toggle) return; + + toggle.addEventListener('change', () => { + if (toggle.checked) { + if (select) select.disabled = false; + this.enablePerformanceOverlay(); + } else { + this.disablePerformanceOverlay(); + if (select) select.disabled = true; + } + }); + + if (select) { + select.addEventListener('change', () => { + if (toggle.checked && this.performanceData) { + this.applyPerformanceOverlay(select.value); + } + }); + } + } + + getSelectedPerformanceMetric() { + const select = document.getElementById('perfMetricSelect'); + return select ? select.value : 'average_latency'; + } + + setPerformanceStatus(message) { + const el = document.getElementById('perfOverlayStatus'); + if (el) el.textContent = message || ''; + } + + async enablePerformanceOverlay() { + this.setPerformanceStatus('Loading performance data...'); + try { + const perfData = await this.fetchPerformanceGraphData(); + this.performanceData = perfData; + this.performanceOverlayActive = true; + this.applyPerformanceOverlay(this.getSelectedPerformanceMetric()); + } catch (error) { + console.warn('Performance overlay unavailable:', error.message); + this.performanceOverlayActive = false; + this.performanceData = null; + this.setPerformanceStatus('Performance data unavailable'); + + const toggle = document.getElementById('perfOverlayToggle'); + if (toggle) toggle.checked = false; + const select = document.getElementById('perfMetricSelect'); + if (select) select.disabled = true; + } + } + + disablePerformanceOverlay() { + if (!this.network) return; + + const nodesDS = this.network.body.data.nodes; + const edgesDS = this.network.body.data.edges; + + const nodeUpdates = []; + this.originalNodeVisuals.forEach((visual, id) => { + if (!nodesDS.get(id)) return; + nodeUpdates.push({ + id, + size: visual.size, + color: visual.color, + borderWidth: visual.borderWidth, + shadow: visual.shadow + }); + }); + if (nodeUpdates.length) nodesDS.update(nodeUpdates); + + const edgeUpdates = []; + this.originalEdgeVisuals.forEach((visual, id) => { + if (!edgesDS.get(id)) return; + edgeUpdates.push({ + id, + width: visual.width, + color: visual.color + }); + }); + if (edgeUpdates.length) edgesDS.update(edgeUpdates); + + this.performanceOverlayActive = false; + this.performanceData = null; + this.setPerformanceStatus(''); + } + + // Builds the list of candidate URLs to try for the performance API. + // /api/performance/* is registered on the API server, which in local + // development runs on a different port (default 8080) than the + // visualization server (default 3000) that serves this page. In + // single-port deployments (e.g. Railway) the relative path works directly. + getPerformanceApiCandidates(path) { + const candidates = []; + + if (window.PERFORMANCE_API_BASE_URL) { + candidates.push(window.PERFORMANCE_API_BASE_URL.replace(/\/$/, '') + path); + } + + candidates.push(path); + + if (window.location.port && window.location.port !== '8080') { + candidates.push(`${window.location.protocol}//${window.location.hostname}:8080${path}`); + } + + return [...new Set(candidates)]; + } + + async fetchPerformanceGraphData() { + const path = '/api/performance/data/graph'; + const candidates = this.getPerformanceApiCandidates(path); + let lastError = null; + + for (const url of candidates) { + try { + const response = await fetch(url); + if (!response.ok) { + lastError = new Error(`HTTP ${response.status} from ${url}`); + continue; + } + + const contentType = response.headers.get('content-type') || ''; + if (!contentType.includes('application/json')) { + lastError = new Error(`Unexpected content-type from ${url}`); + continue; + } + + const payload = await response.json(); + if (!payload || payload.success !== true || !payload.data) { + lastError = new Error(`Unsuccessful response from ${url}`); + continue; + } + + return payload.data; + } catch (error) { + lastError = error; + } + } + + throw lastError || new Error('No performance API endpoint reachable'); + } + + buildPerformanceIndexes(perfData) { + const byId = new Map(); + const byTableName = new Map(); + const byLabel = new Map(); + + (perfData.nodes || []).forEach(node => { + if (node.id !== undefined && node.id !== null) { + byId.set(String(node.id), node); + } + if (node.table_name) { + if (!byTableName.has(node.table_name)) byTableName.set(node.table_name, []); + byTableName.get(node.table_name).push(node); + } + if (node.label) { + if (!byLabel.has(node.label)) byLabel.set(node.label, []); + byLabel.get(node.label).push(node); + } + }); + + return { byId, byTableName, byLabel }; + } + + // Best-effort matching between a domain graph node (/api/graph) and a + // performance graph node (/api/performance/data/graph). Priority: + // 1) explicit "id" property on the domain node, matched against perf node id + // 2) domain node's "table_name" property, matched against perf table_name + // 3) domain node label used as table name (mirrors backend default when no + // table_name property is set) + // 4) domain node label matched uniquely against a single perf node's label + // Returns null (no throw) when no confident match is found. + matchPerformanceNode(domainNode, indexes) { + const props = domainNode.properties || {}; + + if (props.id !== undefined && props.id !== null) { + const match = indexes.byId.get(String(props.id)); + if (match) return match; + } + + if (props.table_name && indexes.byTableName.has(props.table_name)) { + const candidates = indexes.byTableName.get(props.table_name); + if (candidates.length === 1) return candidates[0]; + } + + if (domainNode.label && indexes.byTableName.has(domainNode.label)) { + const candidates = indexes.byTableName.get(domainNode.label); + if (candidates.length === 1) return candidates[0]; + } + + if (domainNode.label && indexes.byLabel.has(domainNode.label)) { + const candidates = indexes.byLabel.get(domainNode.label); + if (candidates.length === 1) return candidates[0]; + } + + return null; + } + + buildNodeIdTranslationMap(domainNodes, indexes) { + const map = new Map(); + domainNodes.forEach(domainNode => { + const perfNode = this.matchPerformanceNode(domainNode, indexes); + if (perfNode) map.set(String(domainNode.id), perfNode); + }); + return map; + } + + buildPerfEdgeIndex(perfEdges) { + const map = new Map(); + (perfEdges || []).forEach(edge => { + if (!edge.source_id || !edge.target_id) return; + map.set(`${edge.source_id}|${edge.target_id}`, edge); + map.set(`${edge.target_id}|${edge.source_id}`, edge); + }); + return map; + } + + matchPerformanceEdge(domainEdge, nodeIdMap, perfEdgeIndex) { + const fromPerf = nodeIdMap.get(String(domainEdge.from)); + const toPerf = nodeIdMap.get(String(domainEdge.to)); + if (!fromPerf || !toPerf) return null; + return perfEdgeIndex.get(`${fromPerf.id}|${toPerf.id}`) || null; + } + + getNodeMetricValue(perfNode, metric) { + const perf = perfNode.performance || {}; + switch (metric) { + case 'queries_per_second': return perf.queries_per_second || 0; + case 'hotspot_score': return perf.hotspot_score || 0; + case 'load_score': return perf.load_score || 0; + case 'average_latency': + default: return perf.average_latency || 0; + } + } + + normalize(value, min, max) { + if (max <= min) return 0; + return Math.min(1, Math.max(0, (value - min) / (max - min))); + } + + scaleRange(normalized, min, max) { + return min + normalized * (max - min); + } + + metricToColor(normalized) { + if (normalized > 0.75) return '#f44336'; + if (normalized > 0.5) return '#ff9800'; + if (normalized > 0.25) return '#ffeb3b'; + return '#4caf50'; + } + + edgeMetricToColor(perf) { + const rankColors = { + critical: '#f44336', + poor: '#f44336', + fair: '#ff9800', + good: '#ffeb3b', + excellent: '#4caf50' + }; + const rank = (perf && perf.performance_rank || '').toLowerCase(); + if (rank && rankColors[rank]) return rankColors[rank]; + + const latency = (perf && perf.average_latency) || 0; + if (latency > 500) return '#f44336'; + if (latency > 200) return '#ff9800'; + if (latency > 100) return '#ffeb3b'; + return '#4caf50'; + } + + applyPerformanceOverlay(metric) { + if (!this.network || !this.performanceData) return; + + const nodesDS = this.network.body.data.nodes; + const edgesDS = this.network.body.data.edges; + + const perfIndexes = this.buildPerformanceIndexes(this.performanceData); + const nodeIdMap = this.buildNodeIdTranslationMap(this.currentGraphNodes, perfIndexes); + const hotspotIds = new Set((this.performanceData.hotspots || []).map(h => h.node_id)); + + const perfNodeValues = (this.performanceData.nodes || []) + .map(node => this.getNodeMetricValue(node, metric)) + .filter(value => typeof value === 'number' && !Number.isNaN(value)); + const minVal = perfNodeValues.length ? Math.min(...perfNodeValues) : 0; + const maxVal = perfNodeValues.length ? Math.max(...perfNodeValues) : 1; + + const nodeUpdates = []; + this.currentGraphNodes.forEach(domainNode => { + const perfNode = nodeIdMap.get(String(domainNode.id)); + if (!perfNode || !this.originalNodeVisuals.has(domainNode.id)) return; + + const value = this.getNodeMetricValue(perfNode, metric); + const normalized = this.normalize(value, minVal, maxVal); + const size = this.scaleRange(normalized, 15, 50); + const color = this.metricToColor(normalized); + const isHotspot = hotspotIds.has(perfNode.id); + + nodeUpdates.push({ + id: domainNode.id, + size, + color: { + background: color, + border: isHotspot ? '#d50000' : color, + highlight: { background: color, border: '#d50000' } + }, + borderWidth: isHotspot ? 4 : 2, + shadow: isHotspot + ? { enabled: true, color: 'rgba(213,0,0,0.6)', size: 15 } + : { enabled: false } + }); + }); + if (nodeUpdates.length) nodesDS.update(nodeUpdates); + + const perfEdgeIndex = this.buildPerfEdgeIndex(this.performanceData.edges); + const freqValues = (this.performanceData.edges || []) + .map(edge => (edge.performance && edge.performance.query_frequency) || 0); + const minFreq = freqValues.length ? Math.min(...freqValues) : 0; + const maxFreq = freqValues.length ? Math.max(...freqValues) : 1; + + const edgeUpdates = []; + this.currentGraphEdges.forEach(domainEdge => { + const perfEdge = this.matchPerformanceEdge(domainEdge, nodeIdMap, perfEdgeIndex); + if (!perfEdge || !this.originalEdgeVisuals.has(domainEdge.id)) return; + + const freq = (perfEdge.performance && perfEdge.performance.query_frequency) || 0; + const width = this.scaleRange(this.normalize(freq, minFreq, maxFreq), 1, 10); + const color = this.edgeMetricToColor(perfEdge.performance); + + edgeUpdates.push({ + id: domainEdge.id, + width, + color: { color, highlight: '#FF6B6B' } + }); + }); + if (edgeUpdates.length) edgesDS.update(edgeUpdates); + + this.setPerformanceStatus( + `Overlay active (${nodeUpdates.length}/${this.currentGraphNodes.length} nodes, ` + + `${edgeUpdates.length}/${this.currentGraphEdges.length} edges matched)` + ); + } } function initThemeToggle() { diff --git a/internal/interfaces/web/templates/visualization.html b/internal/interfaces/web/templates/visualization.html index 4b27f84..2dc0f2e 100644 --- a/internal/interfaces/web/templates/visualization.html +++ b/internal/interfaces/web/templates/visualization.html @@ -19,6 +19,10 @@ .controls .row.g-3 { flex-wrap: wrap; } + + #perfOverlayStatus { + white-space: nowrap; + } @@ -47,6 +51,19 @@ +
+
+ + +
+ + +
@@ -55,7 +72,7 @@
- + diff --git a/tests/integration/performance_schema_test.go b/tests/integration/performance_schema_test.go new file mode 100644 index 0000000..25ceb7b --- /dev/null +++ b/tests/integration/performance_schema_test.go @@ -0,0 +1,78 @@ +/* + * SQL Graph Visualizer - Integration Tests for MySQL Performance Schema collection + * + * Copyright (c) 2025 + * Licensed under Dual License: AGPL-3.0 OR Commercial License + * See LICENSE file for details + */ + +package integration + +import ( + "context" + "database/sql" + "os" + "testing" + "time" + + "sql-graph-visualizer/internal/application/services/performance" + + _ "github.com/go-sql-driver/mysql" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" +) + +// PerformanceSchemaIntegrationTestSuite exercises PerformanceSchemaAdapter +// against the same sakila MySQL test database used by +// DirectDatabaseIntegrationTestSuite (docker-compose service on 127.0.0.1:3308). +type PerformanceSchemaIntegrationTestSuite struct { + suite.Suite + db *sql.DB + adapter *performance.PerformanceSchemaAdapter + ctx context.Context +} + +func (suite *PerformanceSchemaIntegrationTestSuite) SetupSuite() { + suite.ctx = context.Background() + + if os.Getenv("INTEGRATION_TESTS") != "true" { + suite.T().Skip("Integration tests skipped - set INTEGRATION_TESTS=true to enable") + } + + dsn := "sakila_user:sakila123@tcp(127.0.0.1:3308)/sakila" + db, err := sql.Open("mysql", dsn) + require.NoError(suite.T(), err, "should open MySQL connection") + + ctx, cancel := context.WithTimeout(suite.ctx, 5*time.Second) + defer cancel() + require.NoError(suite.T(), db.PingContext(ctx), "test database should be reachable") + + suite.db = db + logger := logrus.New() + logger.SetLevel(logrus.ErrorLevel) + suite.adapter = performance.NewPerformanceSchemaAdapter(db, logger, nil) +} + +func (suite *PerformanceSchemaIntegrationTestSuite) TearDownSuite() { + if suite.db != nil { + _ = suite.db.Close() + } +} + +func (suite *PerformanceSchemaIntegrationTestSuite) TestCollectPerformanceData() { + data, err := suite.adapter.CollectPerformanceData(suite.ctx) + require.NoError(suite.T(), err, "CollectPerformanceData should succeed against a live MySQL instance") + require.NotNil(suite.T(), data) + suite.T().Logf("collected %d statement stats, %d table IO stats", len(data.StatementStats), len(data.TableIOStats)) + + metrics := suite.adapter.ConvertToPerformanceMetrics(data) + require.NotNil(suite.T(), metrics) + + queryPerf := suite.adapter.ConvertToQueryPerformance(data) + suite.T().Logf("converted %d query performance entries", len(queryPerf)) +} + +func TestPerformanceSchemaIntegration(t *testing.T) { + suite.Run(t, new(PerformanceSchemaIntegrationTestSuite)) +}