poimen-workflows/internal/profiling/workflow_profiler.go

package profiling

import (
	"fmt"
	"sort"
	"sync"
)

// TaskProfile represents profiling data for a task
type TaskProfile struct {
	TaskID    string
	Duration  float64
	CPUUsage  float64
	MemUsage  float64
	Throughput float64
}

// WorkflowProfile represents profiling for an entire workflow
type WorkflowProfile struct {
	WorkflowID    string
	Tasks         map[string]*TaskProfile
	TotalDuration float64
	CriticalPath  []string
}

// WorkflowProfiler profiles workflow execution
type WorkflowProfiler struct {
	mu        sync.RWMutex
	profiles  map[string]*WorkflowProfile
}

// NewWorkflowProfiler creates a new workflow profiler
func NewWorkflowProfiler() *WorkflowProfiler {
	return &WorkflowProfiler{
		profiles: make(map[string]*WorkflowProfile),
	}
}

// RecordTaskExecution records task execution metrics
func (wp *WorkflowProfiler) RecordTaskExecution(workflowID, taskID string, duration, cpu, mem float64) {
	wp.mu.Lock()
	defer wp.mu.Unlock()

	if _, exists := wp.profiles[workflowID]; !exists {
		wp.profiles[workflowID] = &WorkflowProfile{
			WorkflowID:   workflowID,
			Tasks:        make(map[string]*TaskProfile),
			CriticalPath: make([]string, 0),
		}
	}

	profile := wp.profiles[workflowID]
	profile.Tasks[taskID] = &TaskProfile{
		TaskID:    taskID,
		Duration:  duration,
		CPUUsage:  cpu,
		MemUsage:  mem,
		Throughput: 1000.0 / duration, // Tasks per second
	}

	// Recalculate total duration
	total := 0.0
	for _, tp := range profile.Tasks {
		if tp.Duration > total {
			total = tp.Duration
		}
	}
	profile.TotalDuration = total
}

// GetSlowTasks returns tasks sorted by duration (slowest first)
func (wp *WorkflowProfiler) GetSlowTasks(workflowID string, limit int) []string {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	if !exists {
		return []string{}
	}

	// Sort tasks by duration
	type taskDuration struct {
		taskID   string
		duration float64
	}

	tasks := make([]taskDuration, 0)
	for taskID, tp := range profile.Tasks {
		tasks = append(tasks, taskDuration{taskID, tp.Duration})
	}

	sort.Slice(tasks, func(i, j int) bool {
		return tasks[i].duration > tasks[j].duration
	})

	result := make([]string, 0)
	for i, t := range tasks {
		if i >= limit {
			break
		}
		result = append(result, t.taskID)
	}

	return result
}

// GetHighCPUTasks returns tasks with high CPU usage
func (wp *WorkflowProfiler) GetHighCPUTasks(workflowID string, threshold float64) []string {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	if !exists {
		return []string{}
	}

	result := make([]string, 0)
	for taskID, tp := range profile.Tasks {
		if tp.CPUUsage > threshold {
			result = append(result, taskID)
		}
	}

	return result
}

// GetHighMemTasks returns tasks with high memory usage
func (wp *WorkflowProfiler) GetHighMemTasks(workflowID string, threshold float64) []string {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	if !exists {
		return []string{}
	}

	result := make([]string, 0)
	for taskID, tp := range profile.Tasks {
		if tp.MemUsage > threshold {
			result = append(result, taskID)
		}
	}

	return result
}

// GetOptimizationSuggestions returns optimization recommendations
func (wp *WorkflowProfiler) GetOptimizationSuggestions(workflowID string) []string {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	if !exists {
		return []string{}
	}

	suggestions := make([]string, 0)

	// Check for slow tasks
	for taskID, tp := range profile.Tasks {
		if tp.Duration > profile.TotalDuration*0.5 {
			suggestions = append(suggestions, fmt.Sprintf("Task %s takes 50%% of total time, consider optimizing", taskID))
		}
		if tp.CPUUsage > 0.8 {
			suggestions = append(suggestions, fmt.Sprintf("Task %s has high CPU usage (%.2f), consider parallelizing", taskID, tp.CPUUsage))
		}
		if tp.MemUsage > 0.8 {
			suggestions = append(suggestions, fmt.Sprintf("Task %s has high memory usage (%.2f), consider reducing payload", taskID, tp.MemUsage))
		}
	}

	return suggestions
}

// GetProfile retrieves profiling data for a workflow
func (wp *WorkflowProfiler) GetProfile(workflowID string) (*WorkflowProfile, bool) {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	return profile, exists
}

// GetTaskProfile retrieves profiling data for a specific task
func (wp *WorkflowProfiler) GetTaskProfile(workflowID, taskID string) (*TaskProfile, error) {
	wp.mu.RLock()
	defer wp.mu.RUnlock()

	profile, exists := wp.profiles[workflowID]
	if !exists {
		return nil, fmt.Errorf("workflow not found: %s", workflowID)
	}

	taskProfile, exists := profile.Tasks[taskID]
	if !exists {
		return nil, fmt.Errorf("task not found: %s", taskID)
	}

	return taskProfile, nil
}

// Clear clears all profiles
func (wp *WorkflowProfiler) Clear() {
	wp.mu.Lock()
	defer wp.mu.Unlock()

	wp.profiles = make(map[string]*WorkflowProfile)
}