From 775ea42b8a05c209f527f9320526f28b1210fec7 Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 16 Jun 2026 01:00:21 +0200 Subject: [PATCH 1/8] benchmark adapter execution; config fix --- internal/application/bootstrap/servers.go | 118 ++++++++++++++++-- .../services/performance/benchmark_service.go | 80 ++++++++++-- .../services/performance/sysbench_adapter.go | 7 ++ internal/domain/models/config.go | 2 +- .../interfaces/api/performance_handlers.go | 76 ++++++++--- 5 files changed, 244 insertions(+), 39 deletions(-) diff --git a/internal/application/bootstrap/servers.go b/internal/application/bootstrap/servers.go index 3ba7ddd..60c0773 100644 --- a/internal/application/bootstrap/servers.go +++ b/internal/application/bootstrap/servers.go @@ -359,7 +359,12 @@ func initPerformanceServices(cfg *models.Config, db *sql.DB) *PerformanceService realtimeMonitor := performance.NewRealtimePerformanceMonitor(logger, realtimeConfig, psAdapter, performanceAnalyzer, graphMapper) 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. benchmarkService := performance.NewBenchmarkService(nil, nil, nil, performanceAnalyzer, logger, benchmarkConfig) + registerBenchmarkTools(benchmarkService, cfg, logger) if cfg.Performance.Realtime != nil && cfg.Performance.Realtime.Enabled { ctx := context.Background() @@ -431,19 +436,108 @@ func createRealtimeConfig(cfg *models.Config) *performance.RealtimeMonitorConfig } func createBenchmarkConfig(cfg *models.Config) *performance.BenchmarkServiceConfig { - config := &performance.BenchmarkServiceConfig{} - if cfg.Performance.Benchmarks != nil { - defaultDuration, _ := time.ParseDuration(cfg.Performance.Benchmarks.DefaultDuration) - maxDuration, _ := time.ParseDuration(cfg.Performance.Benchmarks.MaxDuration) - resultsRetention, _ := time.ParseDuration(cfg.Performance.Benchmarks.ResultsRetention) - config.DefaultTimeout = defaultDuration - config.MaxDuration = maxDuration - config.RetainResults = resultsRetention - config.CleanupInterval = 15 * time.Minute - if cfg.Performance.Benchmarks.Limits != nil { - config.MaxConcurrentRuns = cfg.Performance.Benchmarks.Limits.MaxConcurrentBenchmarks - config.MaxResultsInMemory = cfg.Performance.Benchmarks.Limits.MemoryLimitMB + // Start from sane defaults so safety limits and execution defaults are always + // populated, then override with user configuration where provided. + config := performance.DefaultBenchmarkServiceConfig() + + // Default the benchmark target to the configured source database so that + // benchmark requests work without explicitly specifying a connection. + if url, dbType := benchmarkDatabaseURL(cfg); url != "" { + config.DefaultDatabaseURL = url + config.DefaultDatabaseType = dbType + } + + if b := cfg.Performance.Benchmarks; b != nil { + if d, err := time.ParseDuration(b.DefaultDuration); err == nil && d > 0 { + config.DefaultTestDuration = d + } + if d, err := time.ParseDuration(b.MaxDuration); err == nil && d > 0 { + config.MaxDuration = d + } + if d, err := time.ParseDuration(b.ResultsRetention); err == nil && d > 0 { + config.RetainResults = d + } + if b.Limits != nil { + if b.Limits.MaxConcurrentBenchmarks > 0 { + config.MaxConcurrentRuns = b.Limits.MaxConcurrentBenchmarks + } + if b.Limits.MemoryLimitMB > 0 { + config.MaxResultsInMemory = b.Limits.MemoryLimitMB + } + } + if b.Sysbench != nil && b.Sysbench.Defaults != nil { + if b.Sysbench.Defaults.Threads > 0 { + config.DefaultThreads = b.Sysbench.Defaults.Threads + } + if b.Sysbench.Defaults.TableSize > 0 { + config.DefaultTableSize = b.Sysbench.Defaults.TableSize + } + if b.Sysbench.Defaults.Time > 0 { + config.DefaultTestDuration = time.Duration(b.Sysbench.Defaults.Time) * time.Second + } } } + + // The execution context must comfortably outlast a default-length run. + if config.DefaultTimeout <= config.DefaultTestDuration { + config.DefaultTimeout = config.DefaultTestDuration + 5*time.Minute + } + if config.MaxDuration < config.DefaultTestDuration { + config.MaxDuration = config.DefaultTestDuration + } + 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) { + if b := cfg.Performance.Benchmarks; b != nil && !b.Enabled { + logrus.Info("Benchmarking disabled in configuration; skipping benchmark tool registration") + return + } + + 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) + return + } + logrus.Info("Registered sysbench benchmark tool") +} + +// createSysbenchConfig maps user configuration onto the sysbench adapter config, +// starting from the adapter defaults. +func createSysbenchConfig(cfg *models.Config) *performance.SysbenchConfig { + sbConfig := performance.DefaultSysbenchConfig() + + if b := cfg.Performance.Benchmarks; b != nil && b.Sysbench != nil { + if b.Sysbench.ExecutablePath != "" { + sbConfig.BinaryPath = b.Sysbench.ExecutablePath + } + if d := b.Sysbench.Defaults; d != nil && d.TableSize > 0 { + sbConfig.DefaultTableSize = d.TableSize + } + } + + return sbConfig +} + +// benchmarkDatabaseURL builds a sysbench-compatible database URL and driver type +// from the active source database configuration. +func benchmarkDatabaseURL(cfg *models.Config) (string, string) { + dbCfg := cfg.GetDatabaseConfig() + switch dbCfg.Type { + case models.DatabaseTypePostgreSQL: + if pg := dbCfg.PostgreSQL; pg != nil { + return fmt.Sprintf("postgresql://%s:%s@%s:%d/%s", + pg.GetUsername(), pg.GetPassword(), pg.GetHost(), pg.GetPort(), pg.GetDatabase()), "postgresql" + } + case models.DatabaseTypeMySQL: + if my := dbCfg.MySQL; my != nil { + return fmt.Sprintf("mysql://%s:%s@%s:%d/%s", + my.GetUsername(), my.GetPassword(), my.GetHost(), my.GetPort(), my.GetDatabase()), "mysql" + } + } + return "", "" +} diff --git a/internal/application/services/performance/benchmark_service.go b/internal/application/services/performance/benchmark_service.go index bec4c82..d0a1cab 100644 --- a/internal/application/services/performance/benchmark_service.go +++ b/internal/application/services/performance/benchmark_service.go @@ -50,6 +50,15 @@ type BenchmarkServiceConfig struct { MaxDuration time.Duration `yaml:"max_duration" json:"max_duration"` MaxThreads int `yaml:"max_threads" json:"max_threads"` + // Default execution parameters (applied when a request omits them) + DefaultDatabaseURL string `yaml:"default_database_url" json:"default_database_url"` + DefaultDatabaseType string `yaml:"default_database_type" json:"default_database_type"` + DefaultTestType string `yaml:"default_test_type" json:"default_test_type"` + DefaultTestDuration time.Duration `yaml:"default_test_duration" json:"default_test_duration"` + DefaultThreads int `yaml:"default_threads" json:"default_threads"` + DefaultTables int `yaml:"default_tables" json:"default_tables"` + DefaultTableSize int `yaml:"default_table_size" json:"default_table_size"` + // Tool configurations EnabledTools []string `yaml:"enabled_tools" json:"enabled_tools"` ToolConfigurations map[string]interface{} `yaml:"tool_configurations" json:"tool_configurations"` @@ -142,6 +151,10 @@ func (s *BenchmarkService) GetAvailableTools() []string { // ExecuteBenchmark runs a benchmark with the specified configuration func (s *BenchmarkService) ExecuteBenchmark(ctx context.Context, config ports.BenchmarkConfig, toolName string) (string, error) { + // Fill in any unset fields from the service defaults (database URL, threads, + // table sizing, duration, ...) so callers can submit a minimal request. + s.applyConfigDefaults(&config) + // Validate configuration if err := s.validateConfig(config); err != nil { return "", fmt.Errorf("invalid configuration: %w", err) @@ -163,7 +176,10 @@ func (s *BenchmarkService) ExecuteBenchmark(ctx context.Context, config ports.Be } executionID := uuid.New().String() - executionCtx, cancel := context.WithTimeout(ctx, s.config.DefaultTimeout) + // Detach from the caller's context: benchmarks run asynchronously and must + // outlive the originating HTTP request. Cancellation is handled explicitly + // through CancelBenchmark/StopBenchmark via the stored cancel function. + executionCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.config.DefaultTimeout) execution := &BenchmarkExecution{ ID: executionID, @@ -357,6 +373,33 @@ func (s *BenchmarkService) validateConfig(config ports.BenchmarkConfig) error { return nil } +// applyConfigDefaults fills unset benchmark configuration fields from the +// service defaults so that minimal requests (e.g. only a tool name) still +// produce a runnable configuration. +func (s *BenchmarkService) applyConfigDefaults(config *ports.BenchmarkConfig) { + if config.TestType == "" { + config.TestType = s.config.DefaultTestType + } + if config.DatabaseURL == "" { + config.DatabaseURL = s.config.DefaultDatabaseURL + } + if config.DatabaseType == "" { + config.DatabaseType = s.config.DefaultDatabaseType + } + if config.Threads <= 0 { + config.Threads = s.config.DefaultThreads + } + if config.Tables <= 0 { + config.Tables = s.config.DefaultTables + } + if config.TableSize <= 0 { + config.TableSize = s.config.DefaultTableSize + } + if config.Duration <= 0 { + config.Duration = s.config.DefaultTestDuration + } +} + func (s *BenchmarkService) getBenchmarkTool(name string) (ports.BenchmarkToolPort, error) { s.toolsMutex.RLock() defer s.toolsMutex.RUnlock() @@ -473,17 +516,30 @@ func (s *BenchmarkService) cleanupOldExecutions() { // defaultBenchmarkServiceConfig returns default configuration func defaultBenchmarkServiceConfig() *BenchmarkServiceConfig { return &BenchmarkServiceConfig{ - MaxConcurrentRuns: 5, - DefaultTimeout: 30 * time.Minute, - CleanupInterval: 15 * time.Minute, - RetainResults: 2 * time.Hour, - MaxResultsInMemory: 100, - MaxTableSize: 1000000, // 1M rows - MaxDuration: 1 * time.Hour, - MaxThreads: 64, - EnabledTools: []string{"sysbench", "custom"}, - ToolConfigurations: make(map[string]interface{}), - } + MaxConcurrentRuns: 5, + DefaultTimeout: 30 * time.Minute, + CleanupInterval: 15 * time.Minute, + RetainResults: 2 * time.Hour, + MaxResultsInMemory: 100, + MaxTableSize: 1000000, // 1M rows + MaxDuration: 1 * time.Hour, + MaxThreads: 64, + EnabledTools: []string{"sysbench", "custom"}, + ToolConfigurations: make(map[string]interface{}), + DefaultDatabaseType: "mysql", + DefaultTestType: "oltp_read_write", + DefaultTestDuration: 60 * time.Second, + DefaultThreads: 4, + DefaultTables: 4, + DefaultTableSize: 10000, + } +} + +// DefaultBenchmarkServiceConfig returns the default benchmark service +// configuration. Exported so callers (e.g. bootstrap) can start from sane +// defaults and override individual fields from user configuration. +func DefaultBenchmarkServiceConfig() *BenchmarkServiceConfig { + return defaultBenchmarkServiceConfig() } // Additional methods for integration with existing graph services diff --git a/internal/application/services/performance/sysbench_adapter.go b/internal/application/services/performance/sysbench_adapter.go index f6dd2c8..d48d90d 100644 --- a/internal/application/services/performance/sysbench_adapter.go +++ b/internal/application/services/performance/sysbench_adapter.go @@ -568,6 +568,13 @@ func (s *SysbenchAdapter) classifyPerformanceImpact(avgLatency float64) string { return "HIGH" } +// DefaultSysbenchConfig returns the default sysbench adapter configuration. +// Exported so callers (e.g. bootstrap) can start from sane defaults and +// override individual fields from user configuration. +func DefaultSysbenchConfig() *SysbenchConfig { + return defaultSysbenchConfig() +} + // defaultSysbenchConfig returns default sysbench configuration func defaultSysbenchConfig() *SysbenchConfig { return &SysbenchConfig{ diff --git a/internal/domain/models/config.go b/internal/domain/models/config.go index 6d71510..dc25aa5 100644 --- a/internal/domain/models/config.go +++ b/internal/domain/models/config.go @@ -216,7 +216,7 @@ func (c *Config) GetDatabaseType() DatabaseType { // PerformanceConfig represents the main performance .monitoring configuration type PerformanceConfig struct { - Monitoring *MonitoringConfig `yaml:".monitoring,omitempty"` + Monitoring *MonitoringConfig `yaml:"monitoring,omitempty"` Realtime *RealtimeConfig `yaml:"realtime,omitempty"` Benchmarks *BenchmarksConfig `yaml:"benchmarks,omitempty"` Visualization *VisualizationConfig `yaml:"visualization,omitempty"` diff --git a/internal/interfaces/api/performance_handlers.go b/internal/interfaces/api/performance_handlers.go index 82e99ce..74b3f63 100644 --- a/internal/interfaces/api/performance_handlers.go +++ b/internal/interfaces/api/performance_handlers.go @@ -44,10 +44,22 @@ type Error struct { // BenchmarkRequest represents a benchmark execution request type BenchmarkRequest struct { - BenchmarkType string `json:"benchmark_type"` - Config map[string]interface{} `json:"config"` - Duration int `json:"duration_seconds"` - Description string `json:"description,omitempty"` + // BenchmarkType is kept for backward compatibility. Historically the + // dashboard sends the tool name here (e.g. "sysbench"); it may also carry a + // sysbench test type (e.g. "oltp_read_write"). + BenchmarkType string `json:"benchmark_type"` + Tool string `json:"tool,omitempty"` + TestType string `json:"test_type,omitempty"` + Threads int `json:"threads,omitempty"` + Tables int `json:"tables,omitempty"` + TableSize int `json:"table_size,omitempty"` + WarmupSeconds int `json:"warmup_seconds,omitempty"` + DatabaseURL string `json:"database_url,omitempty"` + DatabaseType string `json:"database_type,omitempty"` + + Config map[string]interface{} `json:"config"` + Duration int `json:"duration_seconds"` + Description string `json:"description,omitempty"` } // BenchmarkStatusResponse represents benchmark status @@ -158,20 +170,25 @@ func (ph *PerformanceHandlers) StartBenchmark(w http.ResponseWriter, r *http.Req return } - // Validate request - if req.BenchmarkType == "" { - ph.sendErrorResponse(w, http.StatusBadRequest, "validation_error", "benchmark_type is required", "") - return - } + // Resolve the tool to run and the sysbench test type. For backward + // compatibility the dashboard sends the tool name in benchmark_type; it may + // also carry a test type. An empty test type lets the service apply its + // configured default. + tool, testType := resolveBenchmarkRequest(req) - // Create benchmark configuration from ports config := ports.BenchmarkConfig{ - TestType: req.BenchmarkType, + TestType: testType, Duration: time.Duration(req.Duration) * time.Second, + Threads: req.Threads, + Tables: req.Tables, + TableSize: req.TableSize, + WarmupTime: time.Duration(req.WarmupSeconds) * time.Second, + DatabaseType: req.DatabaseType, + DatabaseURL: req.DatabaseURL, CustomParams: req.Config, } - executionID, err := ph.benchmarkService.ExecuteBenchmark(r.Context(), config, req.BenchmarkType) + executionID, err := ph.benchmarkService.ExecuteBenchmark(r.Context(), config, tool) if err != nil { ph.sendErrorResponse(w, http.StatusInternalServerError, "benchmark_error", "Failed to start benchmark", err.Error()) return @@ -183,8 +200,9 @@ func (ph *PerformanceHandlers) StartBenchmark(w http.ResponseWriter, r *http.Req StartTime: time.Now(), Progress: 0.0, Metadata: map[string]interface{}{ - "benchmark_type": req.BenchmarkType, - "duration": req.Duration, + "tool": tool, + "test_type": testType, + "duration": req.Duration, }, } @@ -195,6 +213,36 @@ func (ph *PerformanceHandlers) StartBenchmark(w http.ResponseWriter, r *http.Req }) } +// knownBenchmarkTools enumerates tool selectors accepted in the legacy +// benchmark_type field for backward compatibility with the dashboard. +var knownBenchmarkTools = map[string]bool{"sysbench": true, "custom": true} + +// resolveBenchmarkRequest determines the benchmark tool and (optional) sysbench +// test type from a request. It preserves backward compatibility with the +// dashboard, which sends the tool name in benchmark_type. An empty test type is +// returned when none is specified, letting the service apply its default. +func resolveBenchmarkRequest(req BenchmarkRequest) (tool, testType string) { + tool = req.Tool + testType = req.TestType + + switch { + case tool != "": + // explicit tool wins + case knownBenchmarkTools[req.BenchmarkType]: + tool = req.BenchmarkType + case req.BenchmarkType != "": + // benchmark_type carried a test type; default the tool to sysbench + tool = "sysbench" + if testType == "" { + testType = req.BenchmarkType + } + default: + tool = "sysbench" + } + + return tool, testType +} + // GetBenchmark handles requests to get a specific benchmark. func (ph *PerformanceHandlers) GetBenchmark(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) From e34169b0e18f32a59eebf3e24e7060e99a4bd7a7 Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 16 Jun 2026 01:00:50 +0200 Subject: [PATCH 2/8] Makefile - added go clean cache and dep. toolchain --- Makefile | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Makefile b/Makefile index 7b4c0a9..e272843 100644 --- a/Makefile +++ b/Makefile @@ -77,7 +77,9 @@ build-legacy: @echo "Legacy build completed" # Run the application (new CLI) -run: +run: docker-up + @echo "Waiting for Neo4j to be ready..." + @sleep 15 @echo "Starting application..." go run cmd/sql-graph-visualizer/main.go serve @@ -98,7 +100,7 @@ clean: # Start Docker services docker-up: @echo "Starting Docker services..." - docker-compose up -d neo4j-test + docker-compose up -d neo4j-test mysql-test @echo "Docker services started" # Stop Docker services From 1ba82c99e3619c806e3a4fe27a9fab4b7bb3e9db Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Mon, 10 Aug 2026 16:43:45 +0300 Subject: [PATCH 3/8] dependecies update --- go.mod | 10 +++++----- go.sum | 20 ++++++++++---------- 2 files changed, 15 insertions(+), 15 deletions(-) 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= From 50583284a7282ed67e727dbc7f02698e9366702c Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 11 Aug 2026 15:36:30 +0300 Subject: [PATCH 4/8] feat(visualization): add performance overlay toggle to main graph Adds a Performance overlay toggle and metric selector (latency, queries/sec, hotspot score, load score) to the main vis-network graph view. When enabled, it fetches /api/performance/data/graph and maps node/edge performance metrics onto the existing vis-network dataset: node size/color and hotspot border highlighting driven by the selected metric, edge thickness by query_frequency and edge color by latency/ performance_rank. Original node/edge visuals are captured up front and restored exactly when the overlay is disabled. Matching between the domain graph (/api/graph, Neo4j ids) and the performance graph (table/label-derived ids) is best-effort (explicit id property, then table_name, then unique label) and gracefully no-ops for any node/edge that can't be confidently matched, with no thrown JS errors. The fetch also falls back across ports since in local development the visualization server and the API server (which hosts /api/performance/*) listen on different ports; in single-port deployments the relative path is used directly. Only internal/interfaces/web/templates/visualization.html and internal/interfaces/web/static/js/visualization.js were touched; no Go files were modified. Co-Authored-By: Oz --- .../interfaces/web/static/js/visualization.js | 406 +++++++++++++++++- .../web/templates/visualization.html | 19 +- 2 files changed, 423 insertions(+), 2 deletions(-) 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 @@
- + From 461c16910ca92d252b0f3084aadc84bfabff5719 Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 11 Aug 2026 16:14:01 +0300 Subject: [PATCH 5/8] Issue 12 implementation and unit tests --- internal/application/bootstrap/servers.go | 176 ++++++-- .../ports/benchmark_result_store_port.go | 39 ++ .../performance/benchmark_result_store.go | 239 +++++++++++ .../benchmark_result_store_test.go | 174 ++++++++ .../services/performance/benchmark_service.go | 46 +++ .../performance/benchmark_service_test.go | 227 +++++++++++ .../performance/custom_query_adapter.go | 378 ++++++++++++++++++ .../performance/custom_query_adapter_test.go | 260 ++++++++++++ .../performance/graph_performance_mapper.go | 7 + .../graph_performance_mapper_test.go | 65 +++ .../performance/performance_analyzer_test.go | 123 ++++++ .../realtime_performance_monitor.go | 7 + .../performance/sysbench_adapter_test.go | 173 ++++++++ internal/domain/models/config.go | 41 +- .../interfaces/api/performance_handlers.go | 335 +++++++++++++++- .../api/performance_handlers_test.go | 57 +++ 16 files changed, 2282 insertions(+), 65 deletions(-) create mode 100644 internal/application/ports/benchmark_result_store_port.go create mode 100644 internal/application/services/performance/benchmark_result_store.go create mode 100644 internal/application/services/performance/benchmark_result_store_test.go create mode 100644 internal/application/services/performance/benchmark_service_test.go create mode 100644 internal/application/services/performance/custom_query_adapter.go create mode 100644 internal/application/services/performance/custom_query_adapter_test.go create mode 100644 internal/application/services/performance/graph_performance_mapper_test.go create mode 100644 internal/application/services/performance/performance_analyzer_test.go create mode 100644 internal/application/services/performance/sysbench_adapter_test.go create mode 100644 internal/interfaces/api/performance_handlers_test.go diff --git a/internal/application/bootstrap/servers.go b/internal/application/bootstrap/servers.go index 60c0773..8c94515 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,15 @@ 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 +392,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 +421,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 +516,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 +529,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 +608,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) + } + }) + } +} From 3d857e445e3643c0ce1f578adaa225d1026efb4d Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 11 Aug 2026 16:15:57 +0300 Subject: [PATCH 6/8] Makefile - fix Neo4j starting time --- Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 870fd0a18357a4e53abb45d176b39b6a7296c9b3 Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 11 Aug 2026 16:16:19 +0300 Subject: [PATCH 7/8] cmd/main.go - add neo4jRepo --- cmd/main.go | 1 + 1 file changed, 1 insertion(+) 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") From f64a052f4726284ce1fee39094e36c9669b61ec8 Mon Sep 17 00:00:00 2001 From: Petr Stepanek Date: Tue, 11 Aug 2026 16:16:40 +0300 Subject: [PATCH 8/8] tests/integration/performance_schema_test.go - add performance test --- tests/integration/performance_schema_test.go | 78 ++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 tests/integration/performance_schema_test.go 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)) +}