From eead8b2337729a7068740c80ddf20c396bb3514b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 17 Feb 2026 01:38:20 -0800 Subject: [PATCH] feat(plugin): Add job dispatcher for orchestration --- weed/admin/plugin/dispatcher.go | 382 ++++++++++++++++++++++++++++++++ 1 file changed, 382 insertions(+) create mode 100644 weed/admin/plugin/dispatcher.go diff --git a/weed/admin/plugin/dispatcher.go b/weed/admin/plugin/dispatcher.go new file mode 100644 index 000000000..b61bb4ba0 --- /dev/null +++ b/weed/admin/plugin/dispatcher.go @@ -0,0 +1,382 @@ +package plugin + +import ( + "fmt" + "sync" + "time" +) + +// Dispatcher orchestrates job detection scheduling and dispatch +type Dispatcher struct { + mu sync.RWMutex + registry *Registry + queue *JobQueue + detectionSchedules map[string]*DetectionSchedule + jobTypeStateManagement map[string]*JobTypeState + lastDetectionTime map[string]time.Time + detectionConcurrencyLimit map[string]int +} + +// DetectionSchedule holds scheduling information for a detection type +type DetectionSchedule struct { + DetectionType string + Interval time.Duration + LastExecuted time.Time + NextExecutionTime time.Time + ExecutionCount int64 + FailureCount int64 + AverageExecutionMs float64 +} + +// JobTypeState manages state for a specific job type +type JobTypeState struct { + JobType string + mu sync.RWMutex + ActiveCount int + MaxConcurrent int + PendingCount int + CompletedCount int + FailedCount int + LastError string + LastExecutionTime time.Time + AverageExecutionMs float64 + ExecutionHistory []time.Duration + MaxHistorySize int +} + +// NewDispatcher creates a new job dispatcher +func NewDispatcher(registry *Registry, queue *JobQueue) *Dispatcher { + return &Dispatcher{ + registry: registry, + queue: queue, + detectionSchedules: make(map[string]*DetectionSchedule), + jobTypeStateManagement: make(map[string]*JobTypeState), + lastDetectionTime: make(map[string]time.Time), + detectionConcurrencyLimit: make(map[string]int), + } +} + +// RegisterDetectionType registers a detection type with scheduling info +func (d *Dispatcher) RegisterDetectionType(detectionType string, interval time.Duration, maxConcurrent int) error { + d.mu.Lock() + defer d.mu.Unlock() + + if _, exists := d.detectionSchedules[detectionType]; exists { + return fmt.Errorf("detection type %s already registered", detectionType) + } + + d.detectionSchedules[detectionType] = &DetectionSchedule{ + DetectionType: detectionType, + Interval: interval, + NextExecutionTime: time.Now(), + } + + d.detectionConcurrencyLimit[detectionType] = maxConcurrent + d.jobTypeStateManagement[detectionType] = &JobTypeState{ + JobType: detectionType, + MaxConcurrent: maxConcurrent, + MaxHistorySize: 100, + ExecutionHistory: make([]time.Duration, 0, 100), + } + + return nil +} + +// UnregisterDetectionType removes a detection type +func (d *Dispatcher) UnregisterDetectionType(detectionType string) error { + d.mu.Lock() + defer d.mu.Unlock() + + if _, exists := d.detectionSchedules[detectionType]; !exists { + return fmt.Errorf("detection type %s not found", detectionType) + } + + delete(d.detectionSchedules, detectionType) + delete(d.detectionConcurrencyLimit, detectionType) + delete(d.jobTypeStateManagement, detectionType) + delete(d.lastDetectionTime, detectionType) + + return nil +} + +// ScheduleDetections checks and schedules detection jobs that are due +func (d *Dispatcher) ScheduleDetections() []string { + d.mu.Lock() + defer d.mu.Unlock() + + var scheduledJobs []string + now := time.Now() + + for detectionType, schedule := range d.detectionSchedules { + if now.After(schedule.NextExecutionTime) { + // Check if we haven't exceeded concurrency limit + state := d.jobTypeStateManagement[detectionType] + state.mu.RLock() + activeCount := state.ActiveCount + maxConcurrent := state.MaxConcurrent + state.mu.RUnlock() + + if activeCount >= maxConcurrent { + continue // Skip this detection type for now + } + + // Create and enqueue job + jobID := fmt.Sprintf("det-%s-%d", detectionType, now.UnixNano()) + job := &Job{ + ID: jobID, + Type: detectionType, + State: JobStatePending, + CreatedAt: now, + } + + if err := d.queue.Enqueue(job); err != nil { + continue + } + + // Update schedule + schedule.NextExecutionTime = now.Add(schedule.Interval) + schedule.ExecutionCount++ + d.lastDetectionTime[detectionType] = now + + scheduledJobs = append(scheduledJobs, jobID) + + // Update state + state.mu.Lock() + state.PendingCount++ + state.mu.Unlock() + } + } + + return scheduledJobs +} + +// DispatchJob assigns a job to an available plugin +func (d *Dispatcher) DispatchJob(job *Job) (string, error) { + d.mu.RLock() + defer d.mu.RUnlock() + + // Find plugins capable of handling this job type + plugins := d.registry.GetPluginsByCapability(job.Type) + if len(plugins) == 0 { + return "", fmt.Errorf("no plugins available for job type %s", job.Type) + } + + // Find least loaded available plugin + var selectedPlugin *ConnectedPlugin + minLoad := int(^uint32(0) >> 1) + + for _, plugin := range plugins { + if plugin.IsHealthy(30 * time.Second) { + plugin.mu.RLock() + if plugin.ActiveJobs < plugin.MaxConcurrentJobs && plugin.ActiveJobs < minLoad { + selectedPlugin = plugin + minLoad = plugin.ActiveJobs + } + plugin.mu.RUnlock() + } + } + + if selectedPlugin == nil { + return "", fmt.Errorf("no healthy plugins available for job type %s", job.Type) + } + + // Assign job to plugin + job.PluginID = selectedPlugin.ID + job.SetState(JobStateScheduled) + selectedPlugin.IncActiveJobs() + + // Update job type state + state := d.jobTypeStateManagement[job.Type] + state.mu.Lock() + state.ActiveCount++ + state.PendingCount-- + state.mu.Unlock() + + return selectedPlugin.ID, nil +} + +// CompleteJob marks a job as completed +func (d *Dispatcher) CompleteJob(job *Job, result *JobResult) error { + d.mu.Lock() + defer d.mu.Unlock() + + job.Result = result + job.SetState(JobStateCompleted) + + // Update plugin + if plugin, err := d.registry.GetPlugin(job.PluginID); err == nil { + plugin.DecActiveJobs() + } + + // Update state + if state, exists := d.jobTypeStateManagement[job.Type]; exists { + state.mu.Lock() + state.ActiveCount-- + state.CompletedCount++ + if job.ExecutionTime > 0 { + state.ExecutionHistory = append(state.ExecutionHistory, job.ExecutionTime) + if len(state.ExecutionHistory) > state.MaxHistorySize { + state.ExecutionHistory = state.ExecutionHistory[1:] + } + d.updateAverageExecutionTime(state) + } + state.LastExecutionTime = time.Now() + state.mu.Unlock() + } + + // Update detection schedule if applicable + if schedule, exists := d.detectionSchedules[job.Type]; exists { + schedule.LastExecuted = time.Now() + } + + // Record execution + record := &ExecutionRecord{ + JobID: job.ID, + JobType: job.Type, + PluginID: job.PluginID, + State: job.State, + CreatedAt: job.CreatedAt, + StartedAt: job.StartedAt, + CompletedAt: job.CompletedAt, + Result: result, + } + d.queue.RecordExecution(record) + + return nil +} + +// FailJob marks a job as failed +func (d *Dispatcher) FailJob(job *Job, errorMsg string) error { + d.mu.Lock() + defer d.mu.Unlock() + + job.LastError = errorMsg + job.SetState(JobStateFailed) + + // Update plugin + if plugin, err := d.registry.GetPlugin(job.PluginID); err == nil { + plugin.DecActiveJobs() + } + + // Update state + if state, exists := d.jobTypeStateManagement[job.Type]; exists { + state.mu.Lock() + state.ActiveCount-- + state.FailedCount++ + state.LastError = errorMsg + state.LastExecutionTime = time.Now() + state.mu.Unlock() + } + + // Update detection schedule + if schedule, exists := d.detectionSchedules[job.Type]; exists { + schedule.FailureCount++ + schedule.LastExecuted = time.Now() + } + + // Record execution + record := &ExecutionRecord{ + JobID: job.ID, + JobType: job.Type, + PluginID: job.PluginID, + State: job.State, + CreatedAt: job.CreatedAt, + StartedAt: job.StartedAt, + CompletedAt: job.CompletedAt, + LastError: errorMsg, + } + d.queue.RecordExecution(record) + + return nil +} + +// updateAverageExecutionTime recalculates average execution time from history +func (d *Dispatcher) updateAverageExecutionTime(state *JobTypeState) { + if len(state.ExecutionHistory) == 0 { + state.AverageExecutionMs = 0 + return + } + + var total int64 + for _, duration := range state.ExecutionHistory { + total += duration.Milliseconds() + } + state.AverageExecutionMs = float64(total) / float64(len(state.ExecutionHistory)) +} + +// GetJobTypeState returns the state for a specific job type +func (d *Dispatcher) GetJobTypeState(jobType string) *JobTypeState { + d.mu.RLock() + defer d.mu.RUnlock() + + if state, exists := d.jobTypeStateManagement[jobType]; exists { + return state + } + return nil +} + +// GetAllJobTypeStates returns all job type states +func (d *Dispatcher) GetAllJobTypeStates() map[string]*JobTypeState { + d.mu.RLock() + defer d.mu.RUnlock() + + result := make(map[string]*JobTypeState) + for jobType, state := range d.jobTypeStateManagement { + result[jobType] = state + } + return result +} + +// GetDetectionSchedule returns the schedule for a detection type +func (d *Dispatcher) GetDetectionSchedule(detectionType string) *DetectionSchedule { + d.mu.RLock() + defer d.mu.RUnlock() + + if schedule, exists := d.detectionSchedules[detectionType]; exists { + return schedule + } + return nil +} + +// GetDueDetections returns all detection types that are due for execution +func (d *Dispatcher) GetDueDetections() []string { + d.mu.RLock() + defer d.mu.RUnlock() + + var due []string + now := time.Now() + + for detectionType, schedule := range d.detectionSchedules { + if now.After(schedule.NextExecutionTime) { + due = append(due, detectionType) + } + } + + return due +} + +// GetDispatcherStats returns overall dispatcher statistics +func (d *Dispatcher) GetDispatcherStats() map[string]interface{} { + d.mu.RLock() + defer d.mu.RUnlock() + + totalActive := 0 + totalCompleted := 0 + totalFailed := 0 + + for _, state := range d.jobTypeStateManagement { + state.mu.RLock() + totalActive += state.ActiveCount + totalCompleted += state.CompletedCount + totalFailed += state.FailedCount + state.mu.RUnlock() + } + + return map[string]interface{}{ + "detection_types_registered": len(d.detectionSchedules), + "total_active_jobs": totalActive, + "total_completed_jobs": totalCompleted, + "total_failed_jobs": totalFailed, + "job_type_states": len(d.jobTypeStateManagement), + } +}