flamenco/tests/performance/load_test.go

package performance_test

// SPDX-License-Identifier: GPL-3.0-or-later

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"runtime"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"

	"projects.blender.org/studio/flamenco/pkg/api"
	"projects.blender.org/studio/flamenco/tests/helpers"
)

// LoadTestSuite provides comprehensive performance testing
type LoadTestSuite struct {
	suite.Suite
	testHelper *helpers.TestHelper
	baseURL    string
	client     *http.Client
}

// LoadTestMetrics tracks performance metrics during testing
type LoadTestMetrics struct {
	TotalRequests    int64
	SuccessfulReqs   int64
	FailedRequests   int64
	TotalLatency     time.Duration
	MinLatency       time.Duration
	MaxLatency       time.Duration
	StartTime        time.Time
	EndTime          time.Time
	RequestsPerSec   float64
	AvgLatency       time.Duration
	ResponseCodes    map[int]int64
	mutex            sync.RWMutex
}

// WorkerSimulator simulates worker behavior for performance testing
type WorkerSimulator struct {
	ID          string
	UUID        string
	client      *http.Client
	baseURL     string
	isActive    int32
	tasksRun    int64
	lastSeen    time.Time
}

// SetupSuite initializes the performance test environment
func (suite *LoadTestSuite) SetupSuite() {
	suite.testHelper = helpers.NewTestHelper(suite.T())
	
	// Use performance-optimized test server
	server := suite.testHelper.StartTestServer()
	suite.baseURL = server.URL
	
	// Configure HTTP client for performance testing
	suite.client = &http.Client{
		Timeout: 30 * time.Second,
		Transport: &http.Transport{
			MaxIdleConns:       100,
			MaxIdleConnsPerHost: 100,
			IdleConnTimeout:    90 * time.Second,
		},
	}
}

// TearDownSuite cleans up the performance test environment
func (suite *LoadTestSuite) TearDownSuite() {
	if suite.testHelper != nil {
		suite.testHelper.Cleanup()
	}
}

// TestConcurrentJobSubmission tests job submission under load
func (suite *LoadTestSuite) TestConcurrentJobSubmission() {
	const (
		numJobs        = 50
		concurrency    = 10
		targetRPS      = 20 // Target requests per second
		maxLatencyMs   = 1000 // Maximum acceptable latency in milliseconds
	)
	
	metrics := &LoadTestMetrics{
		StartTime:     time.Now(),
		ResponseCodes: make(map[int]int64),
		MinLatency:    time.Hour, // Start with very high value
	}
	
	jobChan := make(chan int, numJobs)
	var wg sync.WaitGroup
	
	// Generate job indices
	for i := 0; i < numJobs; i++ {
		jobChan <- i
	}
	close(jobChan)
	
	// Start concurrent workers
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			
			for jobIndex := range jobChan {
				startTime := time.Now()
				
				job := suite.createLoadTestJob(fmt.Sprintf("Load Test Job %d", jobIndex))
				statusCode, err := suite.submitJobForLoad(job)
				
				latency := time.Since(startTime)
				suite.updateMetrics(metrics, statusCode, latency, err)
				
				// Rate limiting to prevent overwhelming the server
				time.Sleep(time.Millisecond * 50)
			}
		}(i)
	}
	
	wg.Wait()
	metrics.EndTime = time.Now()
	
	suite.calculateFinalMetrics(metrics)
	suite.validatePerformanceMetrics(metrics, targetRPS, maxLatencyMs)
	suite.logPerformanceResults("Job Submission Load Test", metrics)
}

// TestMultiWorkerSimulation tests system with multiple active workers
func (suite *LoadTestSuite) TestMultiWorkerSimulation() {
	const (
		numWorkers      = 10
		simulationTime  = 30 * time.Second
		taskRequestRate = time.Second * 2
	)
	
	metrics := &LoadTestMetrics{
		StartTime:     time.Now(),
		ResponseCodes: make(map[int]int64),
		MinLatency:    time.Hour,
	}
	
	// Register workers
	workers := make([]*WorkerSimulator, numWorkers)
	for i := 0; i < numWorkers; i++ {
		worker := suite.createWorkerSimulator(fmt.Sprintf("load-test-worker-%d", i))
		workers[i] = worker
		
		// Register worker
		err := suite.registerWorkerForLoad(worker)
		require.NoError(suite.T(), err, "Failed to register worker %s", worker.ID)
	}
	
	// Submit jobs to create work
	for i := 0; i < 5; i++ {
		job := suite.createLoadTestJob(fmt.Sprintf("Multi-Worker Test Job %d", i))
		_, err := suite.submitJobForLoad(job)
		require.NoError(suite.T(), err)
	}
	
	// Start worker simulation
	ctx, cancel := context.WithTimeout(context.Background(), simulationTime)
	defer cancel()
	
	var wg sync.WaitGroup
	for _, worker := range workers {
		wg.Add(1)
		go func(w *WorkerSimulator) {
			defer wg.Done()
			suite.simulateWorker(ctx, w, metrics, taskRequestRate)
		}(worker)
	}
	
	wg.Wait()
	metrics.EndTime = time.Now()
	
	suite.calculateFinalMetrics(metrics)
	suite.logPerformanceResults("Multi-Worker Simulation", metrics)
	
	// Validate worker performance
	totalTasksRun := int64(0)
	for _, worker := range workers {
		tasksRun := atomic.LoadInt64(&worker.tasksRun)
		totalTasksRun += tasksRun
		suite.T().Logf("Worker %s processed %d tasks", worker.ID, tasksRun)
	}
	
	assert.Greater(suite.T(), totalTasksRun, int64(0), "Workers should have processed some tasks")
}

// TestDatabaseConcurrency tests database operations under concurrent load
func (suite *LoadTestSuite) TestDatabaseConcurrency() {
	const (
		numOperations = 100
		concurrency   = 20
	)
	
	metrics := &LoadTestMetrics{
		StartTime:     time.Now(),
		ResponseCodes: make(map[int]int64),
		MinLatency:    time.Hour,
	}
	
	// Submit initial jobs for testing
	jobIDs := make([]string, 10)
	for i := 0; i < 10; i++ {
		job := suite.createLoadTestJob(fmt.Sprintf("DB Test Job %d", i))
		jobData, _ := json.Marshal(job)
		
		resp, err := suite.makeRequest("POST", "/api/v3/jobs", bytes.NewReader(jobData))
		require.NoError(suite.T(), err)
		require.Equal(suite.T(), http.StatusOK, resp.StatusCode)
		
		var submittedJob api.Job
		json.NewDecoder(resp.Body).Decode(&submittedJob)
		resp.Body.Close()
		jobIDs[i] = submittedJob.Id
	}
	
	operationChan := make(chan int, numOperations)
	var wg sync.WaitGroup
	
	// Generate operations
	for i := 0; i < numOperations; i++ {
		operationChan <- i
	}
	close(operationChan)
	
	// Start concurrent database operations
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			
			for range operationChan {
				startTime := time.Now()
				
				// Mix of read and write operations
				operations := []func() (int, error){
					func() (int, error) { return suite.queryJobsForLoad() },
					func() (int, error) { return suite.queryWorkersForLoad() },
					func() (int, error) { return suite.getJobDetailsForLoad(jobIDs) },
				}
				
				operation := operations[time.Now().UnixNano()%int64(len(operations))]
				statusCode, err := operation()
				
				latency := time.Since(startTime)
				suite.updateMetrics(metrics, statusCode, latency, err)
			}
		}()
	}
	
	wg.Wait()
	metrics.EndTime = time.Now()
	
	suite.calculateFinalMetrics(metrics)
	suite.validateDatabasePerformance(metrics)
	suite.logPerformanceResults("Database Concurrency Test", metrics)
}

// TestMemoryUsageUnderLoad tests memory consumption during high load
func (suite *LoadTestSuite) TestMemoryUsageUnderLoad() {
	const testDuration = 30 * time.Second
	
	// Baseline memory usage
	var baselineStats, peakStats runtime.MemStats
	runtime.GC()
	runtime.ReadMemStats(&baselineStats)
	
	suite.T().Logf("Baseline memory: Alloc=%d KB, TotalAlloc=%d KB, Sys=%d KB", 
		baselineStats.Alloc/1024, baselineStats.TotalAlloc/1024, baselineStats.Sys/1024)
	
	ctx, cancel := context.WithTimeout(context.Background(), testDuration)
	defer cancel()
	
	var wg sync.WaitGroup
	
	// Continuous job submission
	wg.Add(1)
	go func() {
		defer wg.Done()
		jobCount := 0
		
		for {
			select {
			case <-ctx.Done():
				return
			default:
				job := suite.createLoadTestJob(fmt.Sprintf("Memory Test Job %d", jobCount))
				suite.submitJobForLoad(job)
				jobCount++
				time.Sleep(time.Millisecond * 100)
			}
		}
	}()
	
	// Memory monitoring
	wg.Add(1)
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(time.Second)
		defer ticker.Stop()
		
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				var currentStats runtime.MemStats
				runtime.ReadMemStats(&currentStats)
				
				if currentStats.Alloc > peakStats.Alloc {
					peakStats = currentStats
				}
			}
		}
	}()
	
	wg.Wait()
	
	// Final memory check
	runtime.GC()
	var finalStats runtime.MemStats
	runtime.ReadMemStats(&finalStats)
	
	suite.T().Logf("Peak memory: Alloc=%d KB, TotalAlloc=%d KB, Sys=%d KB", 
		peakStats.Alloc/1024, peakStats.TotalAlloc/1024, peakStats.Sys/1024)
	suite.T().Logf("Final memory: Alloc=%d KB, TotalAlloc=%d KB, Sys=%d KB", 
		finalStats.Alloc/1024, finalStats.TotalAlloc/1024, finalStats.Sys/1024)
	
	// Validate memory usage isn't excessive
	memoryGrowth := float64(peakStats.Alloc-baselineStats.Alloc) / float64(baselineStats.Alloc)
	suite.T().Logf("Memory growth: %.2f%%", memoryGrowth*100)
	
	// Memory growth should be reasonable (less than 500%)
	assert.Less(suite.T(), memoryGrowth, 5.0, "Memory growth should be less than 500%")
}

// Helper methods for performance testing

func (suite *LoadTestSuite) createLoadTestJob(name string) api.SubmittedJob {
	return api.SubmittedJob{
		Name:              name,
		Type:              "simple-blender-render",
		Priority:          50,
		SubmitterPlatform: "linux",
		Settings: map[string]interface{}{
			"filepath":               "/shared-storage/test.blend",
			"chunk_size":             1,
			"format":                 "PNG",
			"image_file_extension":   ".png",
			"frames":                 "1-5", // Small frame range for performance testing
		},
	}
}

func (suite *LoadTestSuite) createWorkerSimulator(name string) *WorkerSimulator {
	return &WorkerSimulator{
		ID:      name,
		client:  suite.client,
		baseURL: suite.baseURL,
	}
}

func (suite *LoadTestSuite) submitJobForLoad(job api.SubmittedJob) (int, error) {
	jobData, err := json.Marshal(job)
	if err != nil {
		return 0, err
	}
	
	resp, err := suite.makeRequest("POST", "/api/v3/jobs", bytes.NewReader(jobData))
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	
	return resp.StatusCode, nil
}

func (suite *LoadTestSuite) registerWorkerForLoad(worker *WorkerSimulator) error {
	workerReg := api.WorkerRegistration{
		Name:               worker.ID,
		Address:            "192.168.1.100",
		Platform:           "linux",
		SoftwareVersion:    "3.0.0",
		SupportedTaskTypes: []string{"blender", "ffmpeg"},
	}
	
	workerData, err := json.Marshal(workerReg)
	if err != nil {
		return err
	}
	
	resp, err := suite.makeRequest("POST", "/api/v3/worker/register-worker", bytes.NewReader(workerData))
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	
	if resp.StatusCode == http.StatusOK {
		var registeredWorker api.RegisteredWorker
		json.NewDecoder(resp.Body).Decode(&registeredWorker)
		worker.UUID = registeredWorker.Uuid
		atomic.StoreInt32(&worker.isActive, 1)
		return nil
	}
	
	return fmt.Errorf("failed to register worker, status: %d", resp.StatusCode)
}

func (suite *LoadTestSuite) simulateWorker(ctx context.Context, worker *WorkerSimulator, metrics *LoadTestMetrics, requestRate time.Duration) {
	// Sign on worker
	signOnData, _ := json.Marshal(api.WorkerSignOn{
		Name:               worker.ID,
		SoftwareVersion:    "3.0.0",
		SupportedTaskTypes: []string{"blender", "ffmpeg"},
	})
	
	signOnURL := fmt.Sprintf("/api/v3/worker/%s/sign-on", worker.UUID)
	resp, err := suite.makeRequest("POST", signOnURL, bytes.NewReader(signOnData))
	if err == nil && resp != nil {
		resp.Body.Close()
	}
	
	ticker := time.NewTicker(requestRate)
	defer ticker.Stop()
	
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			suite.simulateTaskRequest(worker, metrics)
		}
	}
}

func (suite *LoadTestSuite) simulateTaskRequest(worker *WorkerSimulator, metrics *LoadTestMetrics) {
	startTime := time.Now()
	
	taskURL := fmt.Sprintf("/api/v3/worker/%s/task", worker.UUID)
	resp, err := suite.makeRequest("POST", taskURL, nil)
	
	latency := time.Since(startTime)
	worker.lastSeen = time.Now()
	
	if err == nil && resp != nil {
		suite.updateMetrics(metrics, resp.StatusCode, latency, nil)
		
		if resp.StatusCode == http.StatusOK {
			// Simulate task completion
			atomic.AddInt64(&worker.tasksRun, 1)
			
			// Parse assigned task
			var task api.AssignedTask
			json.NewDecoder(resp.Body).Decode(&task)
			resp.Body.Close()
			
			// Simulate task execution time
			time.Sleep(time.Millisecond * 100)
			
			// Send task update
			suite.simulateTaskUpdate(worker, task.Uuid)
		} else {
			resp.Body.Close()
		}
	} else {
		suite.updateMetrics(metrics, 0, latency, err)
	}
}

func (suite *LoadTestSuite) simulateTaskUpdate(worker *WorkerSimulator, taskUUID string) {
	update := api.TaskUpdate{
		TaskProgress: &api.TaskProgress{
			PercentageComplete: 100,
		},
		TaskStatus: api.TaskStatusCompleted,
		Log:        "Task completed successfully",
	}
	
	updateData, _ := json.Marshal(update)
	updateURL := fmt.Sprintf("/api/v3/worker/%s/task/%s", worker.UUID, taskUUID)
	
	resp, err := suite.makeRequest("POST", updateURL, bytes.NewReader(updateData))
	if err == nil && resp != nil {
		resp.Body.Close()
	}
}

func (suite *LoadTestSuite) queryJobsForLoad() (int, error) {
	resp, err := suite.makeRequest("GET", "/api/v3/jobs", nil)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	
	return resp.StatusCode, nil
}

func (suite *LoadTestSuite) queryWorkersForLoad() (int, error) {
	resp, err := suite.makeRequest("GET", "/api/v3/workers", nil)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	
	return resp.StatusCode, nil
}

func (suite *LoadTestSuite) getJobDetailsForLoad(jobIDs []string) (int, error) {
	if len(jobIDs) == 0 {
		return 200, nil
	}
	
	jobID := jobIDs[time.Now().UnixNano()%int64(len(jobIDs))]
	resp, err := suite.makeRequest("GET", fmt.Sprintf("/api/v3/jobs/%s", jobID), nil)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	
	return resp.StatusCode, nil
}

func (suite *LoadTestSuite) makeRequest(method, path string, body io.Reader) (*http.Response, error) {
	url := suite.baseURL + path
	req, err := http.NewRequestWithContext(context.Background(), method, url, body)
	if err != nil {
		return nil, err
	}
	
	req.Header.Set("Content-Type", "application/json")
	return suite.client.Do(req)
}

func (suite *LoadTestSuite) updateMetrics(metrics *LoadTestMetrics, statusCode int, latency time.Duration, err error) {
	metrics.mutex.Lock()
	defer metrics.mutex.Unlock()
	
	atomic.AddInt64(&metrics.TotalRequests, 1)
	
	if err != nil {
		atomic.AddInt64(&metrics.FailedRequests, 1)
	} else {
		atomic.AddInt64(&metrics.SuccessfulReqs, 1)
		metrics.ResponseCodes[statusCode]++
	}
	
	metrics.TotalLatency += latency
	if latency < metrics.MinLatency {
		metrics.MinLatency = latency
	}
	if latency > metrics.MaxLatency {
		metrics.MaxLatency = latency
	}
}

func (suite *LoadTestSuite) calculateFinalMetrics(metrics *LoadTestMetrics) {
	duration := metrics.EndTime.Sub(metrics.StartTime).Seconds()
	metrics.RequestsPerSec = float64(metrics.TotalRequests) / duration
	
	if metrics.TotalRequests > 0 {
		metrics.AvgLatency = time.Duration(int64(metrics.TotalLatency) / metrics.TotalRequests)
	}
}

func (suite *LoadTestSuite) validatePerformanceMetrics(metrics *LoadTestMetrics, targetRPS float64, maxLatencyMs int) {
	// Validate success rate
	successRate := float64(metrics.SuccessfulReqs) / float64(metrics.TotalRequests)
	assert.Greater(suite.T(), successRate, 0.95, "Success rate should be above 95%")
	
	// Validate average latency
	assert.Less(suite.T(), metrics.AvgLatency.Milliseconds(), int64(maxLatencyMs), 
		"Average latency should be under %d ms", maxLatencyMs)
	
	suite.T().Logf("Performance targets - RPS: %.2f (target: %.2f), Avg Latency: %v (max: %d ms)", 
		metrics.RequestsPerSec, targetRPS, metrics.AvgLatency, maxLatencyMs)
}

func (suite *LoadTestSuite) validateDatabasePerformance(metrics *LoadTestMetrics) {
	// Database operations should maintain good performance
	assert.Greater(suite.T(), metrics.RequestsPerSec, 10.0, "Database RPS should be above 10")
	assert.Less(suite.T(), metrics.AvgLatency.Milliseconds(), int64(500), "Database queries should be under 500ms")
}

func (suite *LoadTestSuite) logPerformanceResults(testName string, metrics *LoadTestMetrics) {
	suite.T().Logf("=== %s Results ===", testName)
	suite.T().Logf("Total Requests: %d", metrics.TotalRequests)
	suite.T().Logf("Successful: %d (%.2f%%)", metrics.SuccessfulReqs, 
		float64(metrics.SuccessfulReqs)/float64(metrics.TotalRequests)*100)
	suite.T().Logf("Failed: %d", metrics.FailedRequests)
	suite.T().Logf("Requests/sec: %.2f", metrics.RequestsPerSec)
	suite.T().Logf("Avg Latency: %v", metrics.AvgLatency)
	suite.T().Logf("Min Latency: %v", metrics.MinLatency)
	suite.T().Logf("Max Latency: %v", metrics.MaxLatency)
	suite.T().Logf("Duration: %v", metrics.EndTime.Sub(metrics.StartTime))
	
	suite.T().Logf("Response Codes:")
	for code, count := range metrics.ResponseCodes {
		suite.T().Logf("  %d: %d", code, count)
	}
}

// TestSuite runs all performance tests
func TestLoadSuite(t *testing.T) {
	suite.Run(t, new(LoadTestSuite))
}