Integrate wave-based scaling system with WHOOSH server
- Add scaling system components to server initialization - Register scaling API and assignment broker routes - Start bootstrap pool manager in server lifecycle - Add graceful shutdown for scaling controller - Update API routing to use chi.Router instead of gorilla/mux - Fix Docker API compatibility issues - Configure health gates with placeholder URLs for KACHING and BACKBEAT 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -61,9 +61,15 @@ type Server struct {
|
||||
taskService *tasks.Service
|
||||
giteaIntegration *tasks.GiteaIntegration
|
||||
repoMonitor *monitor.Monitor
|
||||
swarmManager *orchestrator.SwarmManager
|
||||
agentDeployer *orchestrator.AgentDeployer
|
||||
validator *validation.Validator
|
||||
swarmManager *orchestrator.SwarmManager
|
||||
agentDeployer *orchestrator.AgentDeployer
|
||||
scalingController *orchestrator.ScalingController
|
||||
healthGates *orchestrator.HealthGates
|
||||
assignmentBroker *orchestrator.AssignmentBroker
|
||||
bootstrapManager *orchestrator.BootstrapPoolManager
|
||||
metricsCollector *orchestrator.ScalingMetricsCollector
|
||||
scalingAPI *orchestrator.ScalingAPI
|
||||
validator *validation.Validator
|
||||
}
|
||||
|
||||
func NewServer(cfg *config.Config, db *database.DB) (*Server, error) {
|
||||
@@ -84,39 +90,91 @@ func NewServer(cfg *config.Config, db *database.DB) (*Server, error) {
|
||||
// Initialize Docker Swarm orchestrator services conditionally
|
||||
var swarmManager *orchestrator.SwarmManager
|
||||
var agentDeployer *orchestrator.AgentDeployer
|
||||
|
||||
var scalingController *orchestrator.ScalingController
|
||||
var healthGates *orchestrator.HealthGates
|
||||
var assignmentBroker *orchestrator.AssignmentBroker
|
||||
var bootstrapManager *orchestrator.BootstrapPoolManager
|
||||
var metricsCollector *orchestrator.ScalingMetricsCollector
|
||||
var scalingAPI *orchestrator.ScalingAPI
|
||||
|
||||
if cfg.Docker.Enabled {
|
||||
var err error
|
||||
swarmManager, err = orchestrator.NewSwarmManager("", "registry.home.deepblack.cloud")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create swarm manager: %w", err)
|
||||
}
|
||||
|
||||
|
||||
agentDeployer = orchestrator.NewAgentDeployer(swarmManager, db.Pool, "registry.home.deepblack.cloud")
|
||||
|
||||
// Initialize scaling system components
|
||||
log.Info().Msg("🌊 Initializing wave-based scaling system")
|
||||
|
||||
// Initialize health gates for scaling decisions
|
||||
healthGates = orchestrator.NewHealthGates(
|
||||
"http://localhost:8081", // KACHING URL - will be configurable
|
||||
"http://localhost:8082", // BACKBEAT URL - will be configurable
|
||||
"http://localhost:8080", // Self for CHORUS health
|
||||
)
|
||||
|
||||
// Initialize bootstrap pool manager
|
||||
bootstrapConfig := orchestrator.BootstrapPoolConfig{
|
||||
MinPoolSize: 5,
|
||||
MaxPoolSize: 30,
|
||||
HealthCheckInterval: 2 * time.Minute,
|
||||
StaleThreshold: 10 * time.Minute,
|
||||
PreferredRoles: []string{"admin", "coordinator", "stable"},
|
||||
}
|
||||
bootstrapManager = orchestrator.NewBootstrapPoolManager(bootstrapConfig)
|
||||
|
||||
// Initialize assignment broker
|
||||
assignmentBroker = orchestrator.NewAssignmentBroker(bootstrapManager)
|
||||
|
||||
// Initialize metrics collector
|
||||
metricsCollector = orchestrator.NewScalingMetricsCollector(1000) // Keep 1000 operations
|
||||
|
||||
// Initialize scaling controller
|
||||
scalingController = orchestrator.NewScalingController(
|
||||
swarmManager,
|
||||
healthGates,
|
||||
assignmentBroker,
|
||||
bootstrapManager,
|
||||
metricsCollector,
|
||||
)
|
||||
|
||||
// Initialize scaling API
|
||||
scalingAPI = orchestrator.NewScalingAPI(scalingController, metricsCollector)
|
||||
|
||||
log.Info().Msg("✅ Wave-based scaling system initialized")
|
||||
} else {
|
||||
log.Warn().Msg("🐳 Docker integration disabled - council agent deployment unavailable")
|
||||
log.Warn().Msg("🐳 Docker integration disabled - scaling system and council agent deployment unavailable")
|
||||
}
|
||||
|
||||
// Initialize repository monitor with team composer, council composer, and agent deployer
|
||||
repoMonitor := monitor.NewMonitor(db.Pool, cfg.GITEA, teamComposer, councilComposer, agentDeployer)
|
||||
|
||||
s := &Server{
|
||||
config: cfg,
|
||||
db: db,
|
||||
giteaClient: gitea.NewClient(cfg.GITEA),
|
||||
webhookHandler: gitea.NewWebhookHandler(cfg.GITEA.WebhookToken),
|
||||
authMiddleware: auth.NewMiddleware(cfg.Auth.JWTSecret, cfg.Auth.ServiceTokens),
|
||||
rateLimiter: auth.NewRateLimiter(100, time.Minute), // 100 requests per minute per IP
|
||||
p2pDiscovery: p2pDiscovery,
|
||||
agentRegistry: agentRegistry,
|
||||
teamComposer: teamComposer,
|
||||
councilComposer: councilComposer,
|
||||
taskService: taskService,
|
||||
giteaIntegration: giteaIntegration,
|
||||
repoMonitor: repoMonitor,
|
||||
swarmManager: swarmManager,
|
||||
agentDeployer: agentDeployer,
|
||||
validator: validation.NewValidator(),
|
||||
config: cfg,
|
||||
db: db,
|
||||
giteaClient: gitea.NewClient(cfg.GITEA),
|
||||
webhookHandler: gitea.NewWebhookHandler(cfg.GITEA.WebhookToken),
|
||||
authMiddleware: auth.NewMiddleware(cfg.Auth.JWTSecret, cfg.Auth.ServiceTokens),
|
||||
rateLimiter: auth.NewRateLimiter(100, time.Minute), // 100 requests per minute per IP
|
||||
p2pDiscovery: p2pDiscovery,
|
||||
agentRegistry: agentRegistry,
|
||||
teamComposer: teamComposer,
|
||||
councilComposer: councilComposer,
|
||||
taskService: taskService,
|
||||
giteaIntegration: giteaIntegration,
|
||||
repoMonitor: repoMonitor,
|
||||
swarmManager: swarmManager,
|
||||
agentDeployer: agentDeployer,
|
||||
scalingController: scalingController,
|
||||
healthGates: healthGates,
|
||||
assignmentBroker: assignmentBroker,
|
||||
bootstrapManager: bootstrapManager,
|
||||
metricsCollector: metricsCollector,
|
||||
scalingAPI: scalingAPI,
|
||||
validator: validation.NewValidator(),
|
||||
}
|
||||
|
||||
// Initialize BACKBEAT integration if enabled
|
||||
@@ -259,6 +317,19 @@ func (s *Server) setupRoutes() {
|
||||
})
|
||||
})
|
||||
|
||||
// Scaling system endpoints
|
||||
if s.scalingAPI != nil {
|
||||
log.Info().Msg("🌊 Registering wave-based scaling API routes")
|
||||
s.scalingAPI.RegisterRoutes(r)
|
||||
}
|
||||
|
||||
// Assignment broker endpoints (if Docker enabled)
|
||||
if s.assignmentBroker != nil {
|
||||
r.Route("/assignments", func(r chi.Router) {
|
||||
s.assignmentBroker.RegisterRoutes(r)
|
||||
})
|
||||
}
|
||||
|
||||
// BACKBEAT monitoring endpoints
|
||||
r.Route("/backbeat", func(r chi.Router) {
|
||||
r.Get("/status", s.backbeatStatusHandler)
|
||||
@@ -277,6 +348,12 @@ func (s *Server) Start(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Start bootstrap pool manager if available
|
||||
if s.bootstrapManager != nil {
|
||||
log.Info().Msg("🔄 Starting bootstrap pool manager")
|
||||
go s.bootstrapManager.Start(ctx)
|
||||
}
|
||||
|
||||
// Start P2P discovery service
|
||||
if err := s.p2pDiscovery.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start P2P discovery: %w", err)
|
||||
@@ -334,6 +411,15 @@ func (s *Server) Shutdown(ctx context.Context) error {
|
||||
log.Info().Msg("🛑 Repository monitoring service stopped")
|
||||
}
|
||||
|
||||
// Stop scaling controller and related services
|
||||
if s.scalingController != nil {
|
||||
if err := s.scalingController.Close(); err != nil {
|
||||
log.Error().Err(err).Msg("Failed to stop scaling controller")
|
||||
} else {
|
||||
log.Info().Msg("🌊 Wave-based scaling controller stopped")
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.httpServer.Shutdown(ctx); err != nil {
|
||||
return fmt.Errorf("server shutdown failed: %w", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user