admin: add plugin lock coordination

weed/admin/dash/admin_server.go

@@ -98,6 +98,7 @@ type AdminServer struct {
 	// Maintenance system
 	maintenanceManager *maintenance.MaintenanceManager
 	plugin             *adminplugin.Plugin
+	pluginLock         *AdminLockManager
 
 	// Topic retention purger
 	topicRetentionPurger *TopicRetentionPurger
@@ -134,6 +135,8 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 	ctx := context.Background()
 	go masterClient.KeepConnectedToMaster(ctx)
 
+	lockManager := NewAdminLockManager(masterClient, adminLockClientName)
+
 	server := &AdminServer{
 		masterClient:                  masterClient,
 		templateFS:                    templateFS,
@@ -145,6 +148,7 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 		collectionStatsCacheThreshold: defaultStatsCacheTimeout,
 		s3TablesManager:               newS3TablesManager(),
 		icebergPort:                   icebergPort,
+		pluginLock:                    lockManager,
 	}
 
 	// Initialize topic retention purger
@@ -228,6 +232,7 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 		ClusterContextProvider: func(_ context.Context) (*plugin_pb.ClusterContext, error) {
 			return server.buildDefaultPluginClusterContext(), nil
 		},
+		LockManager: lockManager,
 	})
 	if err != nil && dataDir != "" {
 		glog.Warningf("Failed to initialize plugin with dataDir=%q: %v. Falling back to in-memory plugin state.", dataDir, err)
@@ -236,6 +241,7 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
 			ClusterContextProvider: func(_ context.Context) (*plugin_pb.ClusterContext, error) {
 				return server.buildDefaultPluginClusterContext(), nil
 			},
+			LockManager: lockManager,
 		})
 	}
 	if err != nil {
@@ -889,6 +895,13 @@ func (s *AdminServer) GetPlugin() *adminplugin.Plugin {
 	return s.plugin
 }
 
+func (s *AdminServer) acquirePluginLock(reason string) (func(), error) {
+	if s == nil || s.pluginLock == nil {
+		return func() {}, nil
+	}
+	return s.pluginLock.Acquire(reason)
+}
+
 // RequestPluginJobTypeDescriptor asks one worker for job type schema and returns the descriptor.
 func (s *AdminServer) RequestPluginJobTypeDescriptor(ctx context.Context, jobType string, forceRefresh bool) (*plugin_pb.JobTypeDescriptor, error) {
 	if s.plugin == nil {
@@ -931,6 +944,13 @@ func (s *AdminServer) RunPluginDetection(
 	if s.plugin == nil {
 		return nil, fmt.Errorf("plugin is not enabled")
 	}
+	releaseLock, err := s.acquirePluginLock(fmt.Sprintf("plugin detection %s", jobType))
+	if err != nil {
+		return nil, err
+	}
+	if releaseLock != nil {
+		defer releaseLock()
+	}
 	return s.plugin.RunDetection(ctx, jobType, clusterContext, maxResults)
 }
 
@@ -956,6 +976,13 @@ func (s *AdminServer) RunPluginDetectionWithReport(
 	if s.plugin == nil {
 		return nil, fmt.Errorf("plugin is not enabled")
 	}
+	releaseLock, err := s.acquirePluginLock(fmt.Sprintf("plugin detection %s", jobType))
+	if err != nil {
+		return nil, err
+	}
+	if releaseLock != nil {
+		defer releaseLock()
+	}
 	return s.plugin.RunDetectionWithReport(ctx, jobType, clusterContext, maxResults)
 }
 
@@ -969,6 +996,17 @@ func (s *AdminServer) ExecutePluginJob(
 	if s.plugin == nil {
 		return nil, fmt.Errorf("plugin is not enabled")
 	}
+	jobType := ""
+	if job != nil {
+		jobType = strings.TrimSpace(job.JobType)
+	}
+	releaseLock, err := s.acquirePluginLock(fmt.Sprintf("plugin execution %s", jobType))
+	if err != nil {
+		return nil, err
+	}
+	if releaseLock != nil {
+		defer releaseLock()
+	}
 	return s.plugin.ExecuteJob(ctx, job, clusterContext, attempt)
 }