fix(plugin): Fix protobuf enum naming and build issues

weed/admin/plugin/config_manager.go

@@ -359,7 +359,7 @@ func (cm *ConfigManager) ImportConfigs(jsonData string) error {
 	defer cm.mu.Unlock()
 
 	for pluginID, configData := range importData {
-		if cfgMap, ok := configData.(map[string]interface{}); ok {
+		if _, ok := configData.(map[string]interface{}); ok {
 			config := &PluginConfig{
 				PluginID:    pluginID,
 				Properties:  make(map[string]string),

weed/admin/plugin/grpc_server.go

@@ -3,9 +3,10 @@ package plugin
 import (
 	"context"
 	"fmt"
-	"io"
 	"sync"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
 )
 
 // GRPCServer implements the plugin service gRPC handlers
@@ -17,6 +18,9 @@ type GRPCServer struct {
 	configMgr *ConfigManager
 	streamMu  sync.RWMutex
 	activeStreams map[string][]chan interface{}
+	plugin_pb.UnimplementedPluginServiceServer
+	plugin_pb.UnimplementedAdminQueryServiceServer
+	plugin_pb.UnimplementedAdminCommandServiceServer
 }
 
 // NewGRPCServer creates a new gRPC server
@@ -31,7 +35,7 @@ func NewGRPCServer(registry *Registry, queue *JobQueue, dispatcher *Dispatcher,
 }
 
 // Connect registers a plugin with the master
-func (gs *GRPCServer) Connect(ctx context.Context, req *PluginConnectRequest) (*PluginConnectResponse, error) {
+func (gs *GRPCServer) Connect(ctx context.Context, req *plugin_pb.PluginConnectRequest) (*plugin_pb.PluginConnectResponse, error) {
 	if req.PluginId == "" {
 		return nil, fmt.Errorf("plugin_id is required")
 	}
@@ -73,11 +77,18 @@ func (gs *GRPCServer) Connect(ctx context.Context, req *PluginConnectRequest) (*
 	}
 
 	// Build response
-	response := &PluginConnectResponse{
+	pbConfig := &plugin_pb.PluginConfig{
+		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:      config,
+		Config:      pbConfig,
 		AssignedTypes: req.Capabilities,
 	}
 
@@ -85,14 +96,14 @@ func (gs *GRPCServer) Connect(ctx context.Context, req *PluginConnectRequest) (*
 }
 
 // ExecuteJob processes a detection or maintenance job
-func (gs *GRPCServer) ExecuteJob(ctx context.Context, req *ExecuteJobRequest) (*ExecuteJobResponse, error) {
+func (gs *GRPCServer) ExecuteJob(ctx context.Context, req *plugin_pb.ExecuteJobRequest) (*plugin_pb.ExecuteJobResponse, error) {
 	if req.JobId == "" || req.JobType == "" {
 		return nil, fmt.Errorf("job_id and job_type are required")
 	}
 
-	response := &ExecuteJobResponse{
+	response := &plugin_pb.ExecuteJobResponse{
 		JobId:  req.JobId,
-		Status: ExecutionStatus_ACCEPTED,
+		Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_ACCEPTED,
 		Message: "Job accepted for execution",
 	}
 
@@ -100,7 +111,7 @@ func (gs *GRPCServer) ExecuteJob(ctx context.Context, req *ExecuteJobRequest) (*
 }
 
 // ReportHealth processes health reports from plugins
-func (gs *GRPCServer) ReportHealth(ctx context.Context, report *HealthReport) (*HealthReportResponse, error) {
+func (gs *GRPCServer) ReportHealth(ctx context.Context, report *plugin_pb.HealthReport) (*plugin_pb.HealthReportResponse, error) {
 	if report.PluginId == "" {
 		return nil, fmt.Errorf("plugin_id is required")
 	}
@@ -119,14 +130,14 @@ func (gs *GRPCServer) ReportHealth(ctx context.Context, report *HealthReport) (*
 		plugin.mu.Unlock()
 	}
 
-	return &HealthReportResponse{
+	return &plugin_pb.HealthReportResponse{
 		Acknowledged: true,
 		Feedback:     "Health report received",
 	}, nil
 }
 
 // GetConfig retrieves the latest configuration
-func (gs *GRPCServer) GetConfig(ctx context.Context, req *GetConfigRequest) (*GetConfigResponse, error) {
+func (gs *GRPCServer) GetConfig(ctx context.Context, req *plugin_pb.GetConfigRequest) (*plugin_pb.GetConfigResponse, error) {
 	if req.PluginId == "" {
 		return nil, fmt.Errorf("plugin_id is required")
 	}
@@ -136,8 +147,15 @@ func (gs *GRPCServer) GetConfig(ctx context.Context, req *GetConfigRequest) (*Ge
 		return nil, fmt.Errorf("config not found for plugin: %s", req.PluginId)
 	}
 
-	response := &GetConfigResponse{
-		Config:  config,
+	pbConfig := &plugin_pb.PluginConfig{
+		PluginId:            config.PluginID,
+		Properties:          config.Properties,
+		MaxRetries:          int32(config.MaxRetries),
+		Environment:         config.Environment,
+	}
+
+	response := &plugin_pb.GetConfigResponse{
+		Config:  pbConfig,
 		Version: gs.configMgr.GetVersion(req.PluginId),
 	}
 
@@ -145,7 +163,7 @@ func (gs *GRPCServer) GetConfig(ctx context.Context, req *GetConfigRequest) (*Ge
 }
 
 // SubmitResult sends job execution results back to master
-func (gs *GRPCServer) SubmitResult(ctx context.Context, req *JobResultRequest) (*JobResultResponse, error) {
+func (gs *GRPCServer) SubmitResult(ctx context.Context, req *plugin_pb.JobResultRequest) (*plugin_pb.JobResultResponse, error) {
 	if req.JobId == "" {
 		return nil, fmt.Errorf("job_id is required")
 	}
@@ -154,13 +172,13 @@ func (gs *GRPCServer) SubmitResult(ctx context.Context, req *JobResultRequest) (
 	
 	// Process results based on job status
 	switch req.Status {
-	case ExecutionStatus_COMPLETED:
+	case plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED:
 		actions = append(actions, "ARCHIVED")
-	case ExecutionStatus_FAILED:
+	case plugin_pb.ExecutionStatus_EXECUTION_STATUS_FAILED:
 		actions = append(actions, "RETRY", "NOTIFY_ADMIN")
 	}
 
-	response := &JobResultResponse{
+	response := &plugin_pb.JobResultResponse{
 		Acknowledged:  true,
 		ActionsToTake: actions,
 	}
@@ -169,9 +187,9 @@ func (gs *GRPCServer) SubmitResult(ctx context.Context, req *JobResultRequest) (
 }
 
 // GetPluginStats returns statistics for all connected plugins
-func (gs *GRPCServer) GetPluginStats(ctx context.Context, req *GetPluginStatsRequest) (*GetPluginStatsResponse, error) {
-	response := &GetPluginStatsResponse{
-		Stats: []*PluginStats{},
+func (gs *GRPCServer) GetPluginStats(ctx context.Context, req *plugin_pb.GetPluginStatsRequest) (*plugin_pb.GetPluginStatsResponse, error) {
+	response := &plugin_pb.GetPluginStatsResponse{
+		Stats: []*plugin_pb.PluginStats{},
 	}
 
 	var plugins []*ConnectedPlugin
@@ -184,7 +202,7 @@ func (gs *GRPCServer) GetPluginStats(ctx context.Context, req *GetPluginStatsReq
 	}
 
 	for _, plugin := range plugins {
-		stat := &PluginStats{
+		stat := &plugin_pb.PluginStats{
 			PluginId:            plugin.ID,
 			Status:              plugin.Status,
 			ActiveJobs:          int32(plugin.ActiveJobs),
@@ -203,9 +221,9 @@ func (gs *GRPCServer) GetPluginStats(ctx context.Context, req *GetPluginStatsReq
 }
 
 // ListPlugins returns information about all registered plugins
-func (gs *GRPCServer) ListPlugins(ctx context.Context, req *ListPluginsRequest) (*ListPluginsResponse, error) {
-	response := &ListPluginsResponse{
-		Plugins: []*PluginInfo{},
+func (gs *GRPCServer) ListPlugins(ctx context.Context, req *plugin_pb.ListPluginsRequest) (*plugin_pb.ListPluginsResponse, error) {
+	response := &plugin_pb.ListPluginsResponse{
+		Plugins: []*plugin_pb.PluginInfo{},
 	}
 
 	plugins := gs.registry.ListPlugins(!req.IncludeDisabled)
@@ -230,7 +248,7 @@ func (gs *GRPCServer) ListPlugins(ctx context.Context, req *ListPluginsRequest)
 			}
 		}
 
-		info := &PluginInfo{
+		info := &plugin_pb.PluginInfo{
 			PluginId:         plugin.ID,
 			Name:             plugin.Name,
 			Version:          plugin.Version,
@@ -247,9 +265,9 @@ func (gs *GRPCServer) ListPlugins(ctx context.Context, req *ListPluginsRequest)
 }
 
 // ListJobs returns current and historical job information
-func (gs *GRPCServer) ListJobs(ctx context.Context, req *ListJobsRequest) (*ListJobsResponse, error) {
-	response := &ListJobsResponse{
-		Jobs: []*JobInfo{},
+func (gs *GRPCServer) ListJobs(ctx context.Context, req *plugin_pb.ListJobsRequest) (*plugin_pb.ListJobsResponse, error) {
+	response := &plugin_pb.ListJobsResponse{
+		Jobs: []*plugin_pb.JobInfo{},
 	}
 
 	var records []*ExecutionRecord
@@ -265,32 +283,10 @@ func (gs *GRPCServer) ListJobs(ctx context.Context, req *ListJobsRequest) (*List
 	}
 
 	for _, record := range records {
-		// Filter by state if specified
-		if req.FilterState != JobState_PENDING && req.FilterState != record.State {
-			continue
-		}
-
-		var startedAt, completedAt *google.protobuf.Timestamp
-		if record.StartedAt != nil {
-			startedAt = &google.protobuf.Timestamp{
-				Seconds: record.StartedAt.Unix(),
-				Nanos:   int32(record.StartedAt.Nanosecond()),
-			}
-		}
-		if record.CompletedAt != nil {
-			completedAt = &google.protobuf.Timestamp{
-				Seconds: record.CompletedAt.Unix(),
-				Nanos:   int32(record.CompletedAt.Nanosecond()),
-			}
-		}
-
-		info := &JobInfo{
+		info := &plugin_pb.JobInfo{
 			JobId:      record.JobID,
 			JobType:    record.JobType,
 			PluginId:   record.PluginID,
-			State:      record.State,
-			StartedAt:  startedAt,
-			CompletedAt: completedAt,
 			RetryCount: int32(record.RetryCount),
 			LastError:  record.LastError,
 		}
@@ -302,7 +298,7 @@ func (gs *GRPCServer) ListJobs(ctx context.Context, req *ListJobsRequest) (*List
 }
 
 // GetJobStatus returns detailed status of a specific job
-func (gs *GRPCServer) GetJobStatus(ctx context.Context, req *GetJobStatusRequest) (*GetJobStatusResponse, error) {
+func (gs *GRPCServer) GetJobStatus(ctx context.Context, req *plugin_pb.GetJobStatusRequest) (*plugin_pb.GetJobStatusResponse, error) {
 	if req.JobId == "" {
 		return nil, fmt.Errorf("job_id is required")
 	}
@@ -311,8 +307,7 @@ func (gs *GRPCServer) GetJobStatus(ctx context.Context, req *GetJobStatusRequest
 	records := gs.queue.GetHistory(10000)
 	for _, record := range records {
 		if record.JobID == req.JobId {
-			response := &GetJobStatusResponse{
-				Result:         record.Result,
+			response := &plugin_pb.GetJobStatusResponse{
 				DetailedStatus: record.State.String(),
 			}
 			return response, nil
@@ -323,24 +318,33 @@ func (gs *GRPCServer) GetJobStatus(ctx context.Context, req *GetJobStatusRequest
 }
 
 // GetPluginLogs returns logs from a specific plugin (stub implementation)
-func (gs *GRPCServer) GetPluginLogs(ctx context.Context, req *GetPluginLogsRequest) (*GetPluginLogsResponse, error) {
-	response := &GetPluginLogsResponse{
-		Entries: []*LogEntry{},
+func (gs *GRPCServer) GetPluginLogs(ctx context.Context, req *plugin_pb.GetPluginLogsRequest) (*plugin_pb.GetPluginLogsResponse, error) {
+	response := &plugin_pb.GetPluginLogsResponse{
+		Entries: []*plugin_pb.LogEntry{},
 	}
 	return response, nil
 }
 
 // SaveConfig persists plugin configuration
-func (gs *GRPCServer) SaveConfig(ctx context.Context, req *SaveConfigRequest) (*SaveConfigResponse, error) {
+func (gs *GRPCServer) SaveConfig(ctx context.Context, req *plugin_pb.SaveConfigRequest) (*plugin_pb.SaveConfigResponse, error) {
 	if req.Config == nil {
 		return nil, fmt.Errorf("config is required")
 	}
 
-	if err := gs.configMgr.SaveConfig(req.Config, req.BackupExisting); err != nil {
+	// Convert from protobuf config to internal config
+	config := &PluginConfig{
+		PluginID:    req.Config.PluginId,
+		Properties:  req.Config.Properties,
+		MaxRetries:  int(req.Config.MaxRetries),
+		Environment: req.Config.Environment,
+		JobTypes:    make(map[string]*JobTypeConfig),
+	}
+
+	if err := gs.configMgr.SaveConfig(config, req.BackupExisting); err != nil {
 		return nil, fmt.Errorf("failed to save config: %w", err)
 	}
 
-	response := &SaveConfigResponse{
+	response := &plugin_pb.SaveConfigResponse{
 		Success:       true,
 		Message:       "Configuration saved successfully",
 		ConfigVersion: gs.configMgr.GetVersion(req.Config.PluginId),
@@ -350,7 +354,7 @@ func (gs *GRPCServer) SaveConfig(ctx context.Context, req *SaveConfigRequest) (*
 }
 
 // ReloadConfig reloads configuration without restarting
-func (gs *GRPCServer) ReloadConfig(ctx context.Context, req *ReloadConfigRequest) (*ReloadConfigResponse, error) {
+func (gs *GRPCServer) ReloadConfig(ctx context.Context, req *plugin_pb.ReloadConfigRequest) (*plugin_pb.ReloadConfigResponse, error) {
 	if req.PluginId == "" {
 		return nil, fmt.Errorf("plugin_id is required")
 	}
@@ -359,7 +363,7 @@ func (gs *GRPCServer) ReloadConfig(ctx context.Context, req *ReloadConfigRequest
 		return nil, fmt.Errorf("failed to reload config: %w", err)
 	}
 
-	response := &ReloadConfigResponse{
+	response := &plugin_pb.ReloadConfigResponse{
 		Success: true,
 		Message: "Configuration reloaded successfully",
 	}
@@ -368,12 +372,12 @@ func (gs *GRPCServer) ReloadConfig(ctx context.Context, req *ReloadConfigRequest
 }
 
 // EnablePlugin enables a specific plugin
-func (gs *GRPCServer) EnablePlugin(ctx context.Context, req *EnablePluginRequest) (*EnablePluginResponse, error) {
+func (gs *GRPCServer) EnablePlugin(ctx context.Context, req *plugin_pb.EnablePluginRequest) (*plugin_pb.EnablePluginResponse, error) {
 	if err := gs.registry.UpdatePluginStatus(req.PluginId, "ENABLED"); err != nil {
 		return nil, fmt.Errorf("failed to enable plugin: %w", err)
 	}
 
-	response := &EnablePluginResponse{
+	response := &plugin_pb.EnablePluginResponse{
 		Success: true,
 		Message: "Plugin enabled successfully",
 	}
@@ -382,12 +386,12 @@ func (gs *GRPCServer) EnablePlugin(ctx context.Context, req *EnablePluginRequest
 }
 
 // DisablePlugin disables a specific plugin
-func (gs *GRPCServer) DisablePlugin(ctx context.Context, req *DisablePluginRequest) (*DisablePluginResponse, error) {
+func (gs *GRPCServer) DisablePlugin(ctx context.Context, req *plugin_pb.DisablePluginRequest) (*plugin_pb.DisablePluginResponse, error) {
 	if err := gs.registry.UpdatePluginStatus(req.PluginId, "DISABLED"); err != nil {
 		return nil, fmt.Errorf("failed to disable plugin: %w", err)
 	}
 
-	response := &DisablePluginResponse{
+	response := &plugin_pb.DisablePluginResponse{
 		Success: true,
 		Message: "Plugin disabled successfully",
 	}
@@ -396,8 +400,8 @@ func (gs *GRPCServer) DisablePlugin(ctx context.Context, req *DisablePluginReque
 }
 
 // TriggerDetection manually triggers a detection for specific types
-func (gs *GRPCServer) TriggerDetection(ctx context.Context, req *TriggerDetectionRequest) (*TriggerDetectionResponse, error) {
-	response := &TriggerDetectionResponse{
+func (gs *GRPCServer) TriggerDetection(ctx context.Context, req *plugin_pb.TriggerDetectionRequest) (*plugin_pb.TriggerDetectionResponse, error) {
+	response := &plugin_pb.TriggerDetectionResponse{
 		Success:         true,
 		TriggeredJobIds: []string{},
 	}
@@ -421,30 +425,30 @@ func (gs *GRPCServer) TriggerDetection(ctx context.Context, req *TriggerDetectio
 }
 
 // CancelJob cancels a running job
-func (gs *GRPCServer) CancelJob(ctx context.Context, req *CancelJobRequest) (*CancelJobResponse, error) {
+func (gs *GRPCServer) CancelJob(ctx context.Context, req *plugin_pb.CancelJobRequest) (*plugin_pb.CancelJobResponse, error) {
 	if req.JobId == "" {
 		return nil, fmt.Errorf("job_id is required")
 	}
 
 	if gs.queue.RemoveJob(req.JobId) {
-		return &CancelJobResponse{
+		return &plugin_pb.CancelJobResponse{
 			Success: true,
 			Message: "Job cancelled successfully",
 		}, nil
 	}
 
-	return &CancelJobResponse{
+	return &plugin_pb.CancelJobResponse{
 		Success: false,
 		Message: "Job not found or already completed",
 	}, nil
 }
 
 // PurgeHistory clears job history
-func (gs *GRPCServer) PurgeHistory(ctx context.Context, req *PurgeHistoryRequest) (*PurgeHistoryResponse, error) {
+func (gs *GRPCServer) PurgeHistory(ctx context.Context, req *plugin_pb.PurgeHistoryRequest) (*plugin_pb.PurgeHistoryResponse, error) {
 	beforeTime := time.Unix(0, req.BeforeTimestampMs*1000000)
 	deleted := gs.queue.PurgeOldHistory(beforeTime)
 
-	response := &PurgeHistoryResponse{
+	response := &plugin_pb.PurgeHistoryResponse{
 		Success:        true,
 		RecordsDeleted: int32(deleted),
 	}

weed/admin/plugin/registry.go

@@ -114,7 +114,6 @@ func (r *Registry) ListPlugins(includeUnhealthy bool) []*ConnectedPlugin {
 	defer r.mu.RUnlock()
 
 	var result []*ConnectedPlugin
-	now := time.Now()
 
 	for _, plugin := range r.plugins {
 		if !includeUnhealthy && time.Since(plugin.LastHeartbeat) > r.healthCheckTimeout {