diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 33e6b7f9e..e6f354157 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -4,12 +4,15 @@ import ( "context" "fmt" "net/http" + "os" + "path/filepath" "sort" "strings" "time" "github.com/gin-gonic/gin" "github.com/seaweedfs/seaweedfs/weed/admin/maintenance" + "github.com/seaweedfs/seaweedfs/weed/admin/plugin" "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/credential" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -254,10 +257,24 @@ func (s *AdminServer) GetCredentialManager() *credential.CredentialManager { // initPluginManager initializes the plugin manager func (s *AdminServer) initPluginManager(dataDir string) { - // For now, keep pluginManager as interface{} to avoid circular imports - // This will be set later when handlers are initialized - // We'll check if it's nil when needed - glog.V(1).Infof("Plugin manager placeholder initialized") + // Create plugin configuration directory if it doesn't exist + pluginConfigDir := filepath.Join(dataDir, "plugins") + if err := os.MkdirAll(pluginConfigDir, 0755); err != nil { + glog.Warningf("Failed to create plugin config directory: %v", err) + return + } + + // Create plugin manager with default configuration + config := plugin.DefaultManagerConfig(pluginConfigDir) + pm, err := plugin.NewManager(config) + if err != nil { + glog.Warningf("Failed to initialize plugin manager: %v", err) + return + } + + // Store the plugin manager + s.pluginManager = pm + glog.Infof("Plugin manager initialized successfully") } // GetPluginManager returns the plugin manager diff --git a/weed/admin/dash/worker_grpc_server.go b/weed/admin/dash/worker_grpc_server.go index 015bd4215..37dbbfed2 100644 --- a/weed/admin/dash/worker_grpc_server.go +++ b/weed/admin/dash/worker_grpc_server.go @@ -9,8 +9,10 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/admin/maintenance" + "github.com/seaweedfs/seaweedfs/weed/admin/plugin" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" @@ -94,6 +96,19 @@ func (s *WorkerGrpcServer) StartWithTLS(port int) error { worker_pb.RegisterWorkerServiceServer(grpcServer, s) + // Register plugin service if plugin manager is available + if s.adminServer.GetPluginManager() != nil { + // Cast the interface{} to *plugin.Manager + if pm, ok := s.adminServer.GetPluginManager().(*plugin.Manager); ok { + if pluginGrpcServer := pm.GetGRPCServer(); pluginGrpcServer != nil { + plugin_pb.RegisterPluginServiceServer(grpcServer, pluginGrpcServer) + plugin_pb.RegisterAdminQueryServiceServer(grpcServer, pluginGrpcServer) + plugin_pb.RegisterAdminCommandServiceServer(grpcServer, pluginGrpcServer) + glog.Infof("Registered plugin services on worker gRPC server") + } + } + } + s.grpcServer = grpcServer s.listener = listener s.running = true diff --git a/weed/admin/handlers/admin_handlers.go b/weed/admin/handlers/admin_handlers.go index 04c31c060..ff86a9c0e 100644 --- a/weed/admin/handlers/admin_handlers.go +++ b/weed/admin/handlers/admin_handlers.go @@ -8,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/seaweedfs/seaweedfs/weed/admin/dash" + "github.com/seaweedfs/seaweedfs/weed/admin/plugin" "github.com/seaweedfs/seaweedfs/weed/admin/view/app" "github.com/seaweedfs/seaweedfs/weed/admin/view/layout" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -39,14 +40,14 @@ func NewAdminHandlers(adminServer *dash.AdminServer) *AdminHandlers { maintenanceHandlers := NewMaintenanceHandlers(adminServer) mqHandlers := NewMessageQueueHandlers(adminServer) serviceAccountHandlers := NewServiceAccountHandlers(adminServer) - + // Get plugin manager from admin server (may be nil) var pluginMgr interface{} if pm := adminServer.GetPluginManager(); pm != nil { pluginMgr = pm } pluginHandlers := NewPluginHandlers(adminServer, pluginMgr) - + return &AdminHandlers{ adminServer: adminServer, authHandlers: authHandlers, @@ -270,13 +271,13 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, adminUser, pluginApi := api.Group("/plugin") { pluginApi.GET("/list", h.pluginHandlers.ListPluginsAPI) - pluginApi.GET("/jobs/:type", h.pluginHandlers.ListJobsAPI) + pluginApi.GET("/jobs/by-type/:type", h.pluginHandlers.ListJobsAPI) pluginApi.GET("/config/:type", h.pluginHandlers.GetConfigAPI) pluginApi.POST("/config/:type/apply", dash.RequireWriteAccess(), h.pluginHandlers.SaveConfigAPI) pluginApi.GET("/detection/history/:type", h.pluginHandlers.GetDetectionHistoryAPI) pluginApi.GET("/execution/history/:type", h.pluginHandlers.GetExecutionHistoryAPI) - pluginApi.POST("/jobs/:type/trigger-detection", dash.RequireWriteAccess(), h.pluginHandlers.TriggerDetectionAPI) - pluginApi.POST("/jobs/:id/cancel", dash.RequireWriteAccess(), h.pluginHandlers.CancelJobAPI) + pluginApi.POST("/trigger-detection/:type", dash.RequireWriteAccess(), h.pluginHandlers.TriggerDetectionAPI) + pluginApi.POST("/cancel-job/:id", dash.RequireWriteAccess(), h.pluginHandlers.CancelJobAPI) } } } else { @@ -320,6 +321,11 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, adminUser, r.GET("/mq/topics", h.mqHandlers.ShowTopics) r.GET("/mq/topics/:namespace/:topic", h.mqHandlers.ShowTopicDetails) + // Plugin management routes + r.GET("/plugins", h.ShowPlugins) + r.GET("/plugins/jobs/:jobType", h.ShowPluginJobs) + r.GET("/plugins/config/:jobType", h.ShowPluginConfig) + // Maintenance system routes r.GET("/maintenance", h.maintenanceHandlers.ShowMaintenanceQueue) r.GET("/maintenance/workers", h.maintenanceHandlers.ShowMaintenanceWorkers) @@ -450,6 +456,19 @@ func (h *AdminHandlers) SetupRoutes(r *gin.Engine, authRequired bool, adminUser, mqApi.POST("/topics/retention/update", h.mqHandlers.UpdateTopicRetentionAPI) mqApi.POST("/retention/purge", h.adminServer.TriggerTopicRetentionPurgeAPI) } + + // Plugin API routes + pluginApi := api.Group("/plugin") + { + pluginApi.GET("/list", h.pluginHandlers.ListPluginsAPI) + pluginApi.GET("/jobs/by-type/:type", h.pluginHandlers.ListJobsAPI) + pluginApi.GET("/config/:type", h.pluginHandlers.GetConfigAPI) + pluginApi.POST("/config/:type/apply", h.pluginHandlers.SaveConfigAPI) + pluginApi.GET("/detection/history/:type", h.pluginHandlers.GetDetectionHistoryAPI) + pluginApi.GET("/execution/history/:type", h.pluginHandlers.GetExecutionHistoryAPI) + pluginApi.POST("/trigger-detection/:type", h.pluginHandlers.TriggerDetectionAPI) + pluginApi.POST("/cancel-job/:id", h.pluginHandlers.CancelJobAPI) + } } } } @@ -700,19 +719,52 @@ func (h *AdminHandlers) getAdminData(c *gin.Context) dash.AdminData { // ShowPlugins displays the plugins overview page func (h *AdminHandlers) ShowPlugins(c *gin.Context) { - plugins := []interface{}{} + plugins := []map[string]interface{}{} jobTypes := make(map[string]interface{}) - + // Get plugin manager from server if pm := h.adminServer.GetPluginManager(); pm != nil { - // TODO: Get actual plugins from plugin manager + // Cast to *plugin.Manager + if pluginMgr, ok := pm.(*plugin.Manager); ok { + // Get list of connected plugins + connectedPlugins := pluginMgr.ListPlugins(false) + for _, p := range connectedPlugins { + plugins = append(plugins, map[string]interface{}{ + "id": p.ID, + "name": p.Name, + "version": p.Version, + "status": p.Status, + "capabilities": p.Capabilities, + "activeJobs": p.ActiveJobs, + "completedJobs": p.CompletedJobs, + "failedJobs": p.FailedJobs, + "connectedAt": p.ConnectedAt, + "lastHeartbeat": p.LastHeartbeat, + }) + + // Build job types map + for _, cap := range p.Capabilities { + if _, exists := jobTypes[cap]; !exists { + jobTypes[cap] = map[string]interface{}{ + "type": cap, + "description": cap, + "pluginCount": 0, + } + } + // Increment plugin count for this capability + if capData, ok := jobTypes[cap].(map[string]interface{}); ok { + capData["pluginCount"] = capData["pluginCount"].(int) + 1 + } + } + } + } } - + component := app.PluginsOverview(app.PluginsPageData{ Plugins: plugins, JobTypes: jobTypes, }) - + htmlContent := layout.Layout(c, component) htmlContent.Render(c.Request.Context(), c.Writer) } @@ -722,13 +774,13 @@ func (h *AdminHandlers) ShowPluginJobs(c *gin.Context) { jobType := c.Param("jobType") jobs := []interface{}{} stateFilter := c.Query("state") - + component := app.PluginJobsMonitoring(app.PluginJobsPageData{ JobType: jobType, Jobs: jobs, StateFilter: stateFilter, }) - + htmlContent := layout.Layout(c, component) htmlContent.Render(c.Request.Context(), c.Writer) } @@ -740,15 +792,15 @@ func (h *AdminHandlers) ShowPluginConfig(c *gin.Context) { if activeTab == "" { activeTab = "config" } - + component := app.PluginConfiguration(app.PluginConfigPageData{ - JobType: jobType, - Config: app.JobTypeConfig{}, - DetectionHistory: []interface{}{}, - ExecutionHistory: []interface{}{}, - ActiveTab: activeTab, + JobType: jobType, + Config: app.JobTypeConfig{}, + DetectionHistory: []interface{}{}, + ExecutionHistory: []interface{}{}, + ActiveTab: activeTab, }) - + htmlContent := layout.Layout(c, component) htmlContent.Render(c.Request.Context(), c.Writer) } diff --git a/weed/admin/plugin/grpc_server.go b/weed/admin/plugin/grpc_server.go index e401370b6..acf51aaa3 100644 --- a/weed/admin/plugin/grpc_server.go +++ b/weed/admin/plugin/grpc_server.go @@ -11,12 +11,12 @@ import ( // GRPCServer implements the plugin service gRPC handlers type GRPCServer struct { - mu sync.RWMutex - registry *Registry - queue *JobQueue - dispatcher *Dispatcher - configMgr *ConfigManager - streamMu sync.RWMutex + mu sync.RWMutex + registry *Registry + queue *JobQueue + dispatcher *Dispatcher + configMgr *ConfigManager + streamMu sync.RWMutex activeStreams map[string][]chan interface{} plugin_pb.UnimplementedPluginServiceServer plugin_pb.UnimplementedAdminQueryServiceServer @@ -78,17 +78,17 @@ func (gs *GRPCServer) Connect(ctx context.Context, req *plugin_pb.PluginConnectR // Build response pbConfig := &plugin_pb.PluginConfig{ - PluginId: config.PluginID, - Properties: config.Properties, - MaxRetries: int32(config.MaxRetries), - Environment: config.Environment, + PluginId: config.PluginID, + Properties: config.Properties, + MaxRetries: int32(config.MaxRetries), + Environment: config.Environment, } response := &plugin_pb.PluginConnectResponse{ - Success: true, - Message: "Plugin registered successfully", - MasterId: "master-1", - Config: pbConfig, + Success: true, + Message: "Plugin registered successfully", + MasterId: "master-1", + Config: pbConfig, AssignedTypes: req.Capabilities, } @@ -102,8 +102,8 @@ func (gs *GRPCServer) ExecuteJob(ctx context.Context, req *plugin_pb.ExecuteJobR } response := &plugin_pb.ExecuteJobResponse{ - JobId: req.JobId, - Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_ACCEPTED, + JobId: req.JobId, + Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_ACCEPTED, Message: "Job accepted for execution", } @@ -148,10 +148,10 @@ func (gs *GRPCServer) GetConfig(ctx context.Context, req *plugin_pb.GetConfigReq } pbConfig := &plugin_pb.PluginConfig{ - PluginId: config.PluginID, - Properties: config.Properties, - MaxRetries: int32(config.MaxRetries), - Environment: config.Environment, + PluginId: config.PluginID, + Properties: config.Properties, + MaxRetries: int32(config.MaxRetries), + Environment: config.Environment, } response := &plugin_pb.GetConfigResponse{ @@ -169,7 +169,7 @@ func (gs *GRPCServer) SubmitResult(ctx context.Context, req *plugin_pb.JobResult } actions := []string{} - + // Process results based on job status switch req.Status { case plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED: @@ -203,16 +203,16 @@ func (gs *GRPCServer) GetPluginStats(ctx context.Context, req *plugin_pb.GetPlug for _, plugin := range plugins { stat := &plugin_pb.PluginStats{ - PluginId: plugin.ID, - Status: plugin.Status, - ActiveJobs: int32(plugin.ActiveJobs), - CompletedJobs: int32(plugin.CompletedJobs), - FailedJobs: int32(plugin.FailedJobs), - TotalDetections: plugin.TotalDetections, - AvgExecutionTimeMs: float32(plugin.AvgExecutionTimeMs), - CpuUsagePercent: float32(plugin.CPUUsagePercent), - MemoryUsageBytes: plugin.MemoryUsageBytes, - UptimeSeconds: int32(time.Since(plugin.ConnectedAt).Seconds()), + PluginId: plugin.ID, + Status: plugin.Status, + ActiveJobs: int32(plugin.ActiveJobs), + CompletedJobs: int32(plugin.CompletedJobs), + FailedJobs: int32(plugin.FailedJobs), + TotalDetections: plugin.TotalDetections, + AvgExecutionTimeMs: float32(plugin.AvgExecutionTimeMs), + CpuUsagePercent: float32(plugin.CPUUsagePercent), + MemoryUsageBytes: plugin.MemoryUsageBytes, + UptimeSeconds: int32(time.Since(plugin.ConnectedAt).Seconds()), } response.Stats = append(response.Stats, stat) } @@ -249,14 +249,14 @@ func (gs *GRPCServer) ListPlugins(ctx context.Context, req *plugin_pb.ListPlugin } info := &plugin_pb.PluginInfo{ - PluginId: plugin.ID, - Name: plugin.Name, - Version: plugin.Version, - Status: plugin.Status, - Capabilities: plugin.Capabilities, + PluginId: plugin.ID, + Name: plugin.Name, + Version: plugin.Version, + Status: plugin.Status, + Capabilities: plugin.Capabilities, MaxConcurrentJobs: int32(plugin.MaxConcurrentJobs), - ActiveJobs: int32(plugin.ActiveJobs), - Metadata: plugin.Metadata, + ActiveJobs: int32(plugin.ActiveJobs), + Metadata: plugin.Metadata, } response.Plugins = append(response.Plugins, info) } diff --git a/weed/admin/view/app/plugins.templ b/weed/admin/view/app/plugins.templ index 65e2999df..6d4066025 100644 --- a/weed/admin/view/app/plugins.templ +++ b/weed/admin/view/app/plugins.templ @@ -1,8 +1,10 @@ package app +import "fmt" + type PluginsPageData struct { -Plugins []interface{} -JobTypes map[string]interface{} + Plugins []map[string]interface{} + JobTypes map[string]interface{} } templ PluginsOverview(data PluginsPageData) { @@ -26,7 +28,7 @@ templ PluginsOverview(data PluginsPageData) {
Connected Plugins
-
0
+
{ fmt.Sprintf("%d", len(data.Plugins)) }
@@ -49,23 +51,35 @@ if len(data.Plugins) == 0 { - + + + + -for i := 0; i < len(data.Plugins); i++ { +for _, p := range data.Plugins { + + + + - } diff --git a/weed/admin/view/app/plugins_templ.go b/weed/admin/view/app/plugins_templ.go index 86745ceef..76bc0be23 100644 --- a/weed/admin/view/app/plugins_templ.go +++ b/weed/admin/view/app/plugins_templ.go @@ -8,8 +8,10 @@ package app import "github.com/a-h/templ" import templruntime "github.com/a-h/templ/runtime" +import "fmt" + type PluginsPageData struct { - Plugins []interface{} + Plugins []map[string]interface{} JobTypes map[string]interface{} } @@ -34,32 +36,161 @@ func PluginsOverview(data PluginsPageData) templ.Component { templ_7745c5c3_Var1 = templ.NopComponent } ctx = templ.ClearChildren(ctx) - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 1, "

Plugins

Refresh
Connected Plugins
0
Connected Plugins
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 1, "

Plugins

Refresh
Connected Plugins
") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var2 string + templ_7745c5c3_Var2, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.Plugins))) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 31, Col: 59} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var2)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 2, "
Connected Plugins
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } if len(data.Plugins) == 0 { - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 2, "
No plugins connected
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 3, "
No plugins connected
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } } else { - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 3, "
PluginPlugin IDName StatusVersionCapabilities Actions
{ p["id"].(string) }{ p["name"].(string) } -Plugin
-v1.0 +if p["status"].(string) == "CONNECTED" { +Connected +} else { +{ p["status"].(string) } +} +		{ p["version"].(string) } +for _, cap := range p["capabilities"].([]string) { +{ cap } +} -Healthy - -Config +Jobs +Config
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 4, "
PluginStatusActions
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } - for i := 0; i < len(data.Plugins); i++ { - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 4, "") + for _, p := range data.Plugins { + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 5, "") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } } - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 5, "
Plugin IDNameStatusVersionCapabilitiesActions
Plugin
v1.0		HealthyConfig
") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var3 string + templ_7745c5c3_Var3, templ_7745c5c3_Err = templ.JoinStringErrs(p["id"].(string)) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 65, Col: 28} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var3)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 6, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var4 string + templ_7745c5c3_Var4, templ_7745c5c3_Err = templ.JoinStringErrs(p["name"].(string)) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 66, Col: 24} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var4)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + if p["status"].(string) == "CONNECTED" { + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 8, "Connected") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + } else { + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 9, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var5 string + templ_7745c5c3_Var5, templ_7745c5c3_Err = templ.JoinStringErrs(p["status"].(string)) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 71, Col: 53} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var5)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 10, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 11, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var6 string + templ_7745c5c3_Var6, templ_7745c5c3_Err = templ.JoinStringErrs(p["version"].(string)) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 74, Col: 27} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var6)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 12, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + for _, cap := range p["capabilities"].([]string) { + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 13, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + var templ_7745c5c3_Var7 string + templ_7745c5c3_Var7, templ_7745c5c3_Err = templ.JoinStringErrs(cap) + if templ_7745c5c3_Err != nil { + return templ.Error{Err: templ_7745c5c3_Err, FileName: `weed/admin/view/app/plugins.templ`, Line: 77, Col: 38} + } + _, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var7)) + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 14, "") + if templ_7745c5c3_Err != nil { + return templ_7745c5c3_Err + } + } + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 15, "Jobs Config
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 18, "
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } } - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 6, "") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 19, "") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } diff --git a/weed/command/command.go b/weed/command/command.go index b2cef4540..6a54ca74f 100644 --- a/weed/command/command.go +++ b/weed/command/command.go @@ -38,6 +38,7 @@ var Commands = []*Command{ cmdMqBroker, cmdMqKafkaGateway, cmdDB, + cmdPluginWorker, cmdS3, cmdScaffold, cmdServer, diff --git a/weed/command/plugin_worker.go b/weed/command/plugin_worker.go new file mode 100644 index 000000000..366c294d7 --- /dev/null +++ b/weed/command/plugin_worker.go @@ -0,0 +1,313 @@ +package command + +import ( + "context" + "fmt" + "os" + "os/signal" + "path/filepath" + "strings" + "sync" + "syscall" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/grace" + "github.com/seaweedfs/seaweedfs/weed/util/version" + "google.golang.org/grpc" +) + +var cmdPluginWorker = &Command{ + UsageLine: "plugin_worker -admin= [-plugins=] [-workingDir=]", + Short: "start a plugin-based worker using the new plugin system", + Long: `Start a worker using the new plugin system. This worker connects to the admin server +via gRPC and registers capabilities for handling maintenance tasks. + +Supported plugins: erasure_coding, vacuum, balance + +Examples: + weed plugin_worker -admin=localhost:23646 + weed plugin_worker -admin=admin.example.com:23646 + weed plugin_worker -admin=localhost:23646 -plugins=erasure_coding,vacuum + weed plugin_worker -admin=localhost:23646 -workingDir=/tmp/worker + weed plugin_worker -admin=localhost:23646 -debug +`, +} + +var ( + pluginWorkerAdminServer = cmdPluginWorker.Flag.String("admin", "localhost:23646", + "admin server gRPC address (usually admin HTTP port + 10000)") + pluginWorkerPlugins = cmdPluginWorker.Flag.String("plugins", "erasure_coding,vacuum,balance", + "comma-separated list of plugin types to enable") + pluginWorkerWorkingDir = cmdPluginWorker.Flag.String("workingDir", "", + "working directory for the plugin worker") + pluginWorkerMaxConcurrent = cmdPluginWorker.Flag.Int("maxConcurrent", 2, + "maximum number of concurrent jobs") + pluginWorkerDebug = cmdPluginWorker.Flag.Bool("debug", false, + "enable debug logging") + pluginWorkerDebugPort = cmdPluginWorker.Flag.Int("debug.port", 6061, + "http port for debugging") + pluginWorkerTimeout = cmdPluginWorker.Flag.Duration("timeout", 30*time.Second, + "gRPC connection timeout") +) + +func init() { + cmdPluginWorker.Run = runPluginWorker +} + +// GenericPluginWorker is a multi-plugin worker that connects to admin server +type GenericPluginWorker struct { + ID string + AdminServer pb.ServerAddress + Plugins []string + MaxConcurrentJobs int + WorkingDir string + PluginServiceClient plugin_pb.PluginServiceClient + Conn *grpc.ClientConn + Context context.Context + Cancel context.CancelFunc + mu sync.RWMutex + isRunning bool +} + +// NewGenericPluginWorker creates a new generic plugin worker +func NewGenericPluginWorker(adminServer pb.ServerAddress, plugins []string, workingDir string, maxConcurrent int) *GenericPluginWorker { + workerID := fmt.Sprintf("worker-%s-%d", hostname(), time.Now().UnixNano()) + return &GenericPluginWorker{ + ID: string(workerID), + AdminServer: adminServer, + Plugins: plugins, + MaxConcurrentJobs: maxConcurrent, + WorkingDir: workingDir, + } +} + +// Start connects to admin server and registers plugins +func (w *GenericPluginWorker) Start(ctx context.Context) error { + w.mu.Lock() + defer w.mu.Unlock() + + if w.isRunning { + return fmt.Errorf("worker is already running") + } + + w.Context, w.Cancel = context.WithCancel(ctx) + + // Create gRPC connection + dialCtx, cancel := context.WithTimeout(w.Context, *pluginWorkerTimeout) + defer cancel() + + grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.worker") + + conn, err := grpc.DialContext(dialCtx, w.AdminServer.ToGrpcAddress(), grpcDialOption) + if err != nil { + return fmt.Errorf("failed to connect to admin grpc server at %s: %v", w.AdminServer.ToGrpcAddress(), err) + } + + w.Conn = conn + w.PluginServiceClient = plugin_pb.NewPluginServiceClient(conn) + + glog.Infof("Successfully connected to admin grpc server at %s", w.AdminServer) + + // Register plugin with admin server + capabilities := []string{} + for _, plugin := range w.Plugins { + plugin = strings.TrimSpace(plugin) + if plugin != "" { + capabilities = append(capabilities, plugin) + } + } + + connectReq := &plugin_pb.PluginConnectRequest{ + PluginId: w.ID, + PluginName: fmt.Sprintf("GenericWorker-%s", strings.Join(w.Plugins, ",")), + Version: version.VERSION, + Capabilities: capabilities, + MaxConcurrentJobs: int32(w.MaxConcurrentJobs), + SupportsStreaming: false, + Port: 0, + Metadata: map[string]string{ + "working_dir": w.WorkingDir, + }, + } + + connectCtx, cancel := context.WithTimeout(w.Context, 10*time.Second) + defer cancel() + + connectResp, err := w.PluginServiceClient.Connect(connectCtx, connectReq) + if err != nil { + return fmt.Errorf("failed to register with admin server: %v", err) + } + + if !connectResp.Success { + return fmt.Errorf("plugin registration failed: %s", connectResp.Message) + } + + glog.Infof("Plugin registered successfully. Assigned types: %v", connectResp.AssignedTypes) + + w.isRunning = true + + // Start background health reporting + go w.healthReportLoop() + + glog.Infof("Plugin worker started successfully with ID: %s", w.ID) + return nil +} + +// Stop gracefully stops the worker +func (w *GenericPluginWorker) Stop() error { + w.mu.Lock() + defer w.mu.Unlock() + + if !w.isRunning { + return nil + } + + w.isRunning = false + if w.Cancel != nil { + w.Cancel() + } + + if w.Conn != nil { + return w.Conn.Close() + } + + return nil +} + +// healthReportLoop sends periodic health reports to admin server +func (w *GenericPluginWorker) healthReportLoop() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-w.Context.Done(): + return + case <-ticker.C: + w.sendHealthReport() + } + } +} + +// sendHealthReport sends a health report to the admin server +func (w *GenericPluginWorker) sendHealthReport() { + healthReport := &plugin_pb.HealthReport{ + PluginId: w.ID, + TimestampMs: time.Now().UnixMilli(), + Status: plugin_pb.HealthStatus_HEALTH_STATUS_HEALTHY, + ActiveJobs: 0, + CpuPercent: 0, + MemoryBytes: 0, + JobProgress: []*plugin_pb.JobProgress{}, + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := w.PluginServiceClient.ReportHealth(ctx, healthReport) + if err != nil { + glog.Warningf("Failed to send health report: %v", err) + } +} + +// IsRunning checks if the worker is currently running +func (w *GenericPluginWorker) IsRunning() bool { + w.mu.RLock() + defer w.mu.RUnlock() + return w.isRunning +} + +func hostname() string { + hostname, err := os.Hostname() + if err != nil { + hostname = "unknown" + } + return hostname +} + +func runPluginWorker(cmd *Command, args []string) bool { + if *pluginWorkerDebug { + grace.StartDebugServer(*pluginWorkerDebugPort) + } + + util.LoadConfiguration("security", false) + + glog.Infof("Starting plugin worker (v%s)", version.VERSION) + glog.Infof("Admin server: %s", *pluginWorkerAdminServer) + glog.Infof("Plugins: %s", *pluginWorkerPlugins) + + // Parse plugins + plugins := strings.Split(*pluginWorkerPlugins, ",") + validPlugins := []string{} + for _, p := range plugins { + p = strings.TrimSpace(p) + if p != "" { + validPlugins = append(validPlugins, p) + } + } + + if len(validPlugins) == 0 { + glog.Fatalf("No valid plugins specified. Valid options: erasure_coding, vacuum, balance") + return false + } + + // Set up working directory + workingDir := *pluginWorkerWorkingDir + if workingDir == "" { + var err error + workingDir, err = os.Getwd() + if err != nil { + glog.Fatalf("Failed to get current working directory: %v", err) + return false + } + } + + // Create and validate working directory + if err := os.MkdirAll(workingDir, 0755); err != nil { + glog.Fatalf("Failed to create working directory: %v", err) + return false + } + + glog.Infof("Working directory: %s", workingDir) + + // Create plugin-specific subdirectories + for _, plugin := range validPlugins { + pluginDir := filepath.Join(workingDir, plugin) + if err := os.MkdirAll(pluginDir, 0755); err != nil { + glog.Fatalf("Failed to create plugin directory %s: %v", pluginDir, err) + return false + } + } + + // Create and start the worker + adminServerAddress := pb.ServerAddress(*pluginWorkerAdminServer) + worker := NewGenericPluginWorker(adminServerAddress, validPlugins, workingDir, *pluginWorkerMaxConcurrent) + + ctx := context.Background() + if err := worker.Start(ctx); err != nil { + glog.Fatalf("Failed to start plugin worker: %v", err) + return false + } + + // Set up signal handling + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + + glog.Infof("Plugin worker is running. Press Ctrl+C to stop") + + // Wait for shutdown signal + <-sigChan + glog.Infof("Shutdown signal received, stopping plugin worker...") + + if err := worker.Stop(); err != nil { + glog.Errorf("Error stopping plugin worker: %v", err) + } + + glog.Infof("Plugin worker stopped") + return true +}