Resolve "Alpamon Agent Refactoring"

cmd/alpamon/command/root.go

@@ -1,18 +1,21 @@
 package command
 
 import (
-	"context"
 	"fmt"
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/alpacax/alpamon/cmd/alpamon/command/ftp"
 	"github.com/alpacax/alpamon/cmd/alpamon/command/setup"
 	"github.com/alpacax/alpamon/cmd/alpamon/command/tunnel"
+	"github.com/alpacax/alpamon/internal/pool"
+	"github.com/alpacax/alpamon/pkg/agent"
 	"github.com/alpacax/alpamon/pkg/collector"
 	"github.com/alpacax/alpamon/pkg/config"
 	"github.com/alpacax/alpamon/pkg/db"
+	"github.com/alpacax/alpamon/pkg/executor"
 	"github.com/alpacax/alpamon/pkg/logger"
 	"github.com/alpacax/alpamon/pkg/pidfile"
 	"github.com/alpacax/alpamon/pkg/runner"
@@ -42,15 +45,16 @@ func init() {
 }
 
 func runAgent() {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	// Create global context manager for the entire application
+	ctxManager := agent.NewContextManager()
+	ctx := ctxManager.Root()
 
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
 
 	go func() {
 		<-sigChan
-		cancel()
+		ctxManager.Shutdown()
 	}()
 
 	// Logger
@@ -72,65 +76,90 @@ func runAgent() {
 	settings := config.LoadConfig(config.Files(name), wsPath)
 	config.InitSettings(settings)
 
+	// Create global worker pool for the entire application using config settings
+	workerPool := pool.NewPool(settings.PoolMaxWorkers, settings.PoolQueueSize)
+	log.Info().Msgf("Initialized global worker pool with %d workers and queue capacity %d",
+		workerPool.MaxWorkers(), workerPool.QueueCapacity())
+
 	// Session
 	session := scheduler.InitSession()
 	commissioned := session.CheckSession(ctx)
 
-	// Reporter
-	scheduler.StartReporters(session)
+	// Reporter - pass context manager for centralized context management
+	reporters := scheduler.StartReporters(session, ctxManager)
 
-	// Log server
-	logServer := logger.NewLogServer()
+	// Log server - pass worker pool and context manager for connection handling
+	logServer := logger.NewLogServer(workerPool, ctxManager)
 	if logServer != nil {
 		go logServer.StartLogServer()
 	}
 
 	log.Info().Msgf("%s initialized and running.", name)
 
-	// Commit
-	runner.CommitAsync(session, commissioned)
+	// Commit - pass context manager for coordinated lifecycle management
+	runner.CommitAsync(session, commissioned, ctxManager)
 
 	// DB
 	client := db.InitDB()
 
-	// Collector
-	metricCollector := collector.InitCollector(session, client)
+	// Collector - pass context manager for centralized context management
+	metricCollector := collector.InitCollector(session, client, ctxManager)
 	if metricCollector != nil {
 		metricCollector.Start()
 	}
 
-	// Websocket Client (Backhaul - commands, sessions)
-	wsClient := runner.NewWebsocketClient(session)
+	// Websocket Client - pass context manager and worker pool for centralized management
+	wsClient := runner.NewWebsocketClient(session, ctxManager, workerPool)
+
+	// Initialize dispatcher system with callbacks
+	dispatcher, err := executor.InitDispatcher(
+		workerPool,
+		ctxManager,
+		session,
+		wsClient,
+		executor.SystemInfoCallbacks{
+			CommitFunc: runner.CommitSystemInfo,
+			SyncFunc:   runner.SyncSystemInfo,
+		},
+	)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to initialize dispatcher system")
+	}
+
+	wsClient.SetDispatcher(dispatcher)
+	log.Info().Msg("Dispatcher system initialized successfully")
+
 	go wsClient.RunForever(ctx)
 
 	// Control Client (Control - sudo approval)
 	controlClient := runner.NewControlClient()
 	go controlClient.RunForever(ctx)
 
+	// Auth Manager for sudo approval workflow
 	authManager := runner.GetAuthManager(controlClient)
 	go authManager.Start(ctx)
 
 	for {
 		select {
 		case <-ctx.Done():
 			log.Info().Msg("Received termination signal. Shutting down...")
-			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, logServer, pidFilePath)
+			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, workerPool, logServer, reporters, pidFilePath)
 			return
 		case <-wsClient.ShutDownChan:
 			log.Info().Msg("Shutdown command received. Shutting down...")
-			cancel()
-			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, logServer, pidFilePath)
+			ctxManager.Shutdown()
+			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, workerPool, logServer, reporters, pidFilePath)
 			return
 		case <-wsClient.RestartChan:
 			log.Info().Msg("Restart command received. Restarting...")
-			cancel()
-			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, logServer, pidFilePath)
+			ctxManager.Shutdown()
+			gracefulShutdown(metricCollector, wsClient, controlClient, authManager, workerPool, logServer, reporters, pidFilePath)
 			restartAgent()
 			return
 		case <-wsClient.CollectorRestartChan:
 			log.Info().Msg("Collector restart command received. Restarting Collector...")
 			metricCollector.Stop()
-			metricCollector = collector.InitCollector(session, client)
+			metricCollector = collector.InitCollector(session, client, ctxManager)
 			metricCollector.Start()
 		}
 	}
@@ -149,7 +178,7 @@ func restartAgent() {
 	}
 }
 
-func gracefulShutdown(collector *collector.Collector, wsClient *runner.WebsocketClient, controlClient *runner.ControlClient, authManager *runner.AuthManager, logServer *logger.LogServer, pidPath string) {
+func gracefulShutdown(collector *collector.Collector, wsClient *runner.WebsocketClient, controlClient *runner.ControlClient, authManager *runner.AuthManager, workerPool *pool.Pool, logServer *logger.LogServer, reporters *scheduler.ReporterManager, pidPath string) {
 	if collector != nil {
 		collector.Stop()
 	}
@@ -162,6 +191,19 @@ func gracefulShutdown(collector *collector.Collector, wsClient *runner.Websocket
 	if authManager != nil {
 		authManager.Stop()
 	}
+	// Shutdown reporters before worker pool
+	if reporters != nil {
+		if err := reporters.Shutdown(1 * time.Second); err != nil {
+			log.Error().Err(err).Msg("Failed to shutdown reporters gracefully")
+		}
+	}
+	// Shutdown the global worker pool
+	if workerPool != nil {
+		log.Info().Msg("Shutting down global worker pool...")
+		if err := workerPool.Shutdown(1 * time.Second); err != nil {
+			log.Error().Err(err).Msg("Failed to shutdown worker pool gracefully")
+		}
+	}
 	if logServer != nil {
 		logServer.Stop()
 	}

configs/alpamon.conf

@@ -8,4 +8,19 @@ verify = {{.Verify}}
 ca_cert = {{.CACert}}
 
 [logging]
-debug = {{.Debug}}
+debug = {{.Debug}}
+
+[pool]
+# Maximum number of concurrent workers in the global worker pool
+# Default: 20
+# max_workers = 20
+
+# Size of the job queue for the global worker pool
+# Default: 200
+# queue_size = 200
+
+# Default timeout in seconds for pool tasks
+# If not set or commented out, the default value of 30 seconds will be used
+# Set to a positive number to specify timeout in seconds
+# Default: 30
+# default_timeout = 30

internal/pool/pool.go

@@ -0,0 +1,176 @@
+// Package pool provides a true worker pool implementation with job queue and context support.
+package pool
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"runtime/debug"
+	"sync"
+	"time"
+)
+
+// Job represents a unit of work to be executed by the pool
+type Job struct {
+	fn func() error
+}
+
+// Pool represents a pool of workers with a job queue
+type Pool struct {
+	maxWorkers int
+	jobQueue   chan *Job
+	ctx        context.Context
+	cancel     context.CancelFunc
+	wg         sync.WaitGroup
+}
+
+// NewPool creates a new worker pool with job queue
+func NewPool(maxWorkers int, queueSize int) *Pool {
+	if maxWorkers <= 0 {
+		maxWorkers = 10
+	}
+	if queueSize <= 0 {
+		queueSize = maxWorkers * 10 // Default queue size
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	p := &Pool{
+		maxWorkers: maxWorkers,
+		jobQueue:   make(chan *Job, queueSize),
+		ctx:        ctx,
+		cancel:     cancel,
+	}
+
+	// Start worker goroutines
+	p.startWorkers()
+
+	return p
+}
+
+// startWorkers launches the worker goroutines
+func (p *Pool) startWorkers() {
+	for i := 0; i < p.maxWorkers; i++ {
+		p.wg.Add(1)
+		go p.worker()
+	}
+}
+
+// worker is the main loop for each worker goroutine
+func (p *Pool) worker() {
+	defer p.wg.Done()
+
+	for {
+		select {
+		case job, ok := <-p.jobQueue:
+			if !ok {
+				// Job queue is closed
+				return
+			}
+			p.executeJob(job)
+		case <-p.ctx.Done():
+			// Pool is shutting down - drain remaining jobs first
+			for {
+				select {
+				case job, ok := <-p.jobQueue:
+					if !ok {
+						return
+					}
+					p.executeJob(job)
+				default:
+					// No more jobs in queue
+					return
+				}
+			}
+		}
+	}
+}
+
+// executeJob runs a job with panic recovery
+func (p *Pool) executeJob(job *Job) {
+	// Panic recovery
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("recovered from panic in pool worker: %v\nstack: %s", r, debug.Stack())
+		}
+	}()
+
+	// Execute the function
+	err := job.fn()
+	if err != nil {
+		log.Printf("pool worker error: %v", err)
+	}
+}
+
+// Submit adds a job to the queue (non-blocking)
+// Returns error if the queue is full or pool is shutting down
+func (p *Pool) Submit(ctx context.Context, fn func() error) error {
+	// Check if pool is shutting down first
+	select {
+	case <-p.ctx.Done():
+		return fmt.Errorf("pool is shutting down")
+	default:
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-p.ctx.Done():
+		return fmt.Errorf("pool is shutting down")
+	case p.jobQueue <- &Job{fn: fn}:
+		return nil
+	default:
+		// Queue is full
+		return fmt.Errorf("job queue is full")
+	}
+}
+
+// QueueSize returns the current number of jobs in the queue
+func (p *Pool) QueueSize() int {
+	return len(p.jobQueue)
+}
+
+// QueueCapacity returns the maximum queue capacity
+func (p *Pool) QueueCapacity() int {
+	return cap(p.jobQueue)
+}
+
+// MaxWorkers returns the number of workers
+func (p *Pool) MaxWorkers() int {
+	return p.maxWorkers
+}
+
+// Shutdown gracefully shuts down the pool
+func (p *Pool) Shutdown(timeout time.Duration) error {
+	// Signal shutdown to prevent new submissions
+	p.cancel()
+
+	// Don't close the job queue immediately - let workers drain it
+	// Workers will exit when context is cancelled
+
+	// Wait for all workers to complete
+	done := make(chan struct{})
+	go func() {
+		p.wg.Wait()
+		// Close the job queue after all workers have exited
+		close(p.jobQueue)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		return nil
+	case <-time.After(timeout):
+		return fmt.Errorf("shutdown timeout after %v", timeout)
+	}
+}
+
+// IsShuttingDown returns true if the pool is shutting down
+func (p *Pool) IsShuttingDown() bool {
+	select {
+	case <-p.ctx.Done():
+		return true
+	default:
+		return false
+	}
+}