change create rule api to run only on master node to avoid race condition

Signed-off-by: Ruirui Zhang <[email protected]>

Author: ruai0511

modules/autotagging-commons/common/src/main/java/org/opensearch/rule/CreateRuleRequest.java

@@ -8,8 +8,8 @@
 
 package org.opensearch.rule;
 
-import org.opensearch.action.ActionRequest;
 import org.opensearch.action.ActionRequestValidationException;
+import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.rule.autotagging.Rule;
@@ -28,7 +28,7 @@
  * }'
  * @opensearch.experimental
  */
-public class CreateRuleRequest extends ActionRequest {
+public class CreateRuleRequest extends ClusterManagerNodeRequest<CreateRuleRequest> {
     private final Rule rule;
 
     /**

modules/autotagging-commons/common/src/test/java/org/opensearch/rule/autotagging/AutoTaggingRegistryTests.java

@@ -15,9 +15,9 @@
 
 import static org.opensearch.rule.autotagging.AutoTaggingRegistry.MAX_FEATURE_TYPE_NAME_LENGTH;
 import static org.opensearch.rule.autotagging.RuleTests.INVALID_FEATURE;
+import static org.opensearch.rule.utils.RuleTestUtils.FEATURE_TYPE_NAME;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
-import static org.opensearch.rule.utils.RuleTestUtils.FEATURE_TYPE_NAME;
 
 public class AutoTaggingRegistryTests extends OpenSearchTestCase {
 

modules/autotagging-commons/src/main/java/org/opensearch/rule/action/TransportCreateRuleAction.java

@@ -9,46 +9,86 @@
 package org.opensearch.rule.action;
 
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.action.support.HandledTransportAction;
+import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.block.ClusterBlockException;
+import org.opensearch.cluster.block.ClusterBlockLevel;
+import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.rule.CreateRuleRequest;
 import org.opensearch.rule.CreateRuleResponse;
 import org.opensearch.rule.RulePersistenceService;
 import org.opensearch.rule.RulePersistenceServiceRegistry;
-import org.opensearch.tasks.Task;
+import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
+import java.io.IOException;
+
+import static org.opensearch.threadpool.ThreadPool.Names.SAME;
+
 /**
  * Transport action to create Rules
  * @opensearch.experimental
  */
-public class TransportCreateRuleAction extends HandledTransportAction<CreateRuleRequest, CreateRuleResponse> {
+public class TransportCreateRuleAction extends TransportClusterManagerNodeAction<CreateRuleRequest, CreateRuleResponse> {
 
     private final RulePersistenceServiceRegistry rulePersistenceServiceRegistry;
 
     /**
      * Constructor for TransportCreateRuleAction
      *
+     * @param threadPool - {@link ThreadPool} object
      * @param transportService - a {@link TransportService} object
+     * @param clusterService - a {@link ClusterService} object
      * @param actionFilters - a {@link ActionFilters} object
+     * @param indexNameExpressionResolver - {@link IndexNameExpressionResolver} object
      * @param rulePersistenceServiceRegistry - a {@link RulePersistenceServiceRegistry} object
      */
     @Inject
     public TransportCreateRuleAction(
+        ThreadPool threadPool,
         TransportService transportService,
+        ClusterService clusterService,
         ActionFilters actionFilters,
+        IndexNameExpressionResolver indexNameExpressionResolver,
         RulePersistenceServiceRegistry rulePersistenceServiceRegistry
     ) {
-        super(CreateRuleAction.NAME, transportService, actionFilters, CreateRuleRequest::new);
+        super(
+            CreateRuleAction.NAME,
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            CreateRuleRequest::new,
+            indexNameExpressionResolver
+        );
         this.rulePersistenceServiceRegistry = rulePersistenceServiceRegistry;
     }
 
     @Override
-    protected void doExecute(Task task, CreateRuleRequest request, ActionListener<CreateRuleResponse> listener) {
+    protected String executor() {
+        return SAME;
+    }
+
+    @Override
+    protected CreateRuleResponse read(StreamInput in) throws IOException {
+        return new CreateRuleResponse(in);
+    }
+
+    @Override
+    protected void clusterManagerOperation(CreateRuleRequest request, ClusterState state, ActionListener<CreateRuleResponse> listener)
+        throws Exception {
         final RulePersistenceService rulePersistenceService = rulePersistenceServiceRegistry.getRulePersistenceService(
             request.getRule().getFeatureType()
         );
         rulePersistenceService.createRule(request, listener);
     }
+
+    @Override
+    protected ClusterBlockException checkBlock(CreateRuleRequest request, ClusterState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.WRITE);
+    }
 }

modules/autotagging-commons/src/test/java/org/opensearch/rule/InMemoryRuleProcessingServiceTests.java

@@ -136,11 +136,6 @@ public String getName() {
         public Map<String, Attribute> getAllowedAttributesRegistry() {
             return Map.of("test_attribute", TestAttribute.TEST_ATTRIBUTE);
         }
-
-        // @Override
-        // public FeatureValueValidator getFeatureValueValidator() {
-        // return null;
-        // }
     }
 
     public enum TestAttribute implements Attribute {

modules/autotagging-commons/src/test/java/org/opensearch/rule/action/CreateRuleActionTests.java

@@ -0,0 +1,22 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.rule.action;
+
+import org.opensearch.core.common.io.stream.Writeable;
+import org.opensearch.test.OpenSearchTestCase;
+
+public class CreateRuleActionTests extends OpenSearchTestCase {
+    public void testGetName() {
+        assertEquals("cluster:admin/opensearch/rule/_create", CreateRuleAction.NAME);
+    }
+
+    public void testCreateResponseReader() {
+        assertTrue(CreateRuleAction.INSTANCE.getResponseReader() instanceof Writeable.Reader);
+    }
+}

modules/autotagging-commons/src/test/java/org/opensearch/rule/action/TransportCreateRuleActionTests.java

@@ -0,0 +1,58 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.rule.action;
+
+import org.opensearch.action.support.ActionFilters;
+import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.rule.CreateRuleRequest;
+import org.opensearch.rule.RulePersistenceService;
+import org.opensearch.rule.RulePersistenceServiceRegistry;
+import org.opensearch.rule.autotagging.Rule;
+import org.opensearch.test.OpenSearchTestCase;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.TransportService;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TransportCreateRuleActionTests extends OpenSearchTestCase {
+    TransportCreateRuleAction sut;
+
+    public void testExecute() throws Exception {
+        RulePersistenceServiceRegistry rulePersistenceServiceRegistry = mock(RulePersistenceServiceRegistry.class);
+        TransportService transportService = mock(TransportService.class);
+        ActionFilters actionFilters = mock(ActionFilters.class);
+        RulePersistenceService rulePersistenceService = mock(RulePersistenceService.class);
+        CreateRuleRequest createRuleRequest = mock(CreateRuleRequest.class);
+        Rule rule = mock(Rule.class);
+        when(createRuleRequest.getRule()).thenReturn(rule);
+        when(createRuleRequest.getRule().getFeatureType()).thenReturn(null);
+        ThreadPool threadPool = mock(ThreadPool.class);
+        ClusterService clusterService = mock(ClusterService.class);
+        IndexNameExpressionResolver indexNameExpressionResolver = mock(IndexNameExpressionResolver.class);
+
+        when(rulePersistenceServiceRegistry.getRulePersistenceService(any())).thenReturn(rulePersistenceService);
+        doNothing().when(rulePersistenceService).getRule(any(), any());
+        sut = new TransportCreateRuleAction(
+            threadPool,
+            transportService,
+            clusterService,
+            actionFilters,
+            indexNameExpressionResolver,
+            rulePersistenceServiceRegistry
+        );
+        sut.clusterManagerOperation(createRuleRequest, null, null);
+        verify(rulePersistenceService, times(1)).createRule(any(), any());
+    }
+}

plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/WorkloadManagementPlugin.java

@@ -49,7 +49,6 @@
 import org.opensearch.rule.InMemoryRuleProcessingService;
 import org.opensearch.rule.RulePersistenceService;
 import org.opensearch.rule.autotagging.FeatureType;
-import org.opensearch.rule.autotagging.FeatureValueValidator;
 import org.opensearch.rule.service.IndexStoredRulePersistenceService;
 import org.opensearch.rule.spi.RuleFrameworkExtension;
 import org.opensearch.rule.storage.DefaultAttributeValueStore;
@@ -102,14 +101,13 @@ public Collection<Object> createComponents(
         IndexNameExpressionResolver indexNameExpressionResolver,
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
-        FeatureValueValidator validator = WorkloadGroupFeatureValueValidator.getInstance(clusterService);
-        WorkloadGroupFeatureType.initializeFeatureValueValidator(validator);
+        FeatureTypeHolder.featureType = new WorkloadGroupFeatureType(new WorkloadGroupFeatureValueValidator(clusterService));
         RulePersistenceServiceHolder.rulePersistenceService = new IndexStoredRulePersistenceService(
             INDEX_NAME,
             client,
             clusterService,
             MAX_RULES_PER_PAGE,
-            new XContentRuleParser(WorkloadGroupFeatureType.getInstance()),
+            new XContentRuleParser(FeatureTypeHolder.featureType),
             new IndexBasedRuleQueryMapper()
         );
         InMemoryRuleProcessingService ruleProcessingService = new InMemoryRuleProcessingService(
@@ -175,10 +173,14 @@ public Supplier<RulePersistenceService> getRulePersistenceServiceSupplier() {
 
     @Override
     public FeatureType getFeatureType() {
-        return WorkloadGroupFeatureType.getInstance();
+        return FeatureTypeHolder.featureType;
     }
 
     static class RulePersistenceServiceHolder {
         private static RulePersistenceService rulePersistenceService;
     }
+
+    static class FeatureTypeHolder {
+        private static FeatureType featureType;
+    }
 }

plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/rule/WorkloadGroupFeatureType.java

@@ -31,37 +31,15 @@ public class WorkloadGroupFeatureType implements FeatureType {
         RuleAttribute.INDEX_PATTERN
     );
     private final FeatureValueValidator featureValueValidator;
-    private static WorkloadGroupFeatureType instance;
 
     /**
      * constructor for WorkloadGroupFeatureType
      * @param featureValueValidator
      */
-    private WorkloadGroupFeatureType(FeatureValueValidator featureValueValidator) {
+    public WorkloadGroupFeatureType(FeatureValueValidator featureValueValidator) {
         this.featureValueValidator = featureValueValidator;
     }
 
-    /**
-     * Initializes the singleton instance of WorkloadGroupFeatureType.
-     * This method should be called once before calling {@link #getInstance()}.
-     * @param validator the FeatureValueValidator to be used for initialization
-     */
-    public static void initializeFeatureValueValidator(FeatureValueValidator validator) {
-        if (instance == null) {
-            instance = new WorkloadGroupFeatureType(validator);
-        }
-    }
-
-    /**
-     * Returns the singleton instance of {@link WorkloadGroupFeatureType}.
-     */
-    public static WorkloadGroupFeatureType getInstance() {
-        if (instance == null) {
-            throw new IllegalStateException("FeatureValueValidator is not initialized. Call initializeFeatureValueValidator() first.");
-        }
-        return instance;
-    }
-
     @Override
     public String getName() {
         return NAME;

plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/rule/WorkloadGroupFeatureValueValidator.java

@@ -20,33 +20,16 @@
  */
 public class WorkloadGroupFeatureValueValidator implements FeatureValueValidator {
     private final ClusterService clusterService;
-    private static volatile WorkloadGroupFeatureValueValidator instance;
     private final Logger logger = LogManager.getLogger(WorkloadGroupFeatureValueValidator.class);
 
     /**
      * constructor for WorkloadGroupFeatureValueValidator
      * @param clusterService
      */
-    private WorkloadGroupFeatureValueValidator(ClusterService clusterService) {
+    public WorkloadGroupFeatureValueValidator(ClusterService clusterService) {
         this.clusterService = clusterService;
     }
 
-    /**
-     * Returns the singleton instance of {@code WorkloadGroupFeatureValueValidator}, initializing it if necessary.
-     * Uses double-checked locking to ensure thread-safe lazy initialization.
-     * @param clusterService the {@link ClusterService} used to construct the validator if not already initialized
-     */
-    public static WorkloadGroupFeatureValueValidator getInstance(ClusterService clusterService) {
-        if (instance == null) {
-            synchronized (WorkloadGroupFeatureValueValidator.class) {
-                if (instance == null) {
-                    instance = new WorkloadGroupFeatureValueValidator(clusterService);
-                }
-            }
-        }
-        return instance;
-    }
-
     @Override
     public void validate(String featureValue) {
         if (!clusterService.state().metadata().workloadGroups().containsKey(featureValue)) {

plugins/workload-management/src/test/java/org/opensearch/plugin/wlm/rule/WorkloadGroupFeatureTypeTests.java

@@ -8,27 +8,15 @@
 
 package org.opensearch.plugin.wlm.rule;
 
-import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.rule.RuleAttribute;
 import org.opensearch.rule.autotagging.Attribute;
 import org.opensearch.rule.autotagging.AutoTaggingRegistry;
-import org.opensearch.rule.autotagging.FeatureValueValidator;
 import org.opensearch.test.OpenSearchTestCase;
-import org.junit.Before;
 
 import java.util.Map;
 
-import static org.mockito.Mockito.mock;
-
 public class WorkloadGroupFeatureTypeTests extends OpenSearchTestCase {
-    WorkloadGroupFeatureType featureType;
-
-    @Before
-    public void setUpFeatureType() {
-        FeatureValueValidator validator = WorkloadGroupFeatureValueValidator.getInstance(mock(ClusterService.class));
-        WorkloadGroupFeatureType.initializeFeatureValueValidator(validator);
-        featureType = WorkloadGroupFeatureType.getInstance();
-    }
+    WorkloadGroupFeatureType featureType = new WorkloadGroupFeatureType(new WorkloadGroupFeatureValueValidator(null));
 
     public void testGetName_returnsCorrectName() {
         assertEquals("workload_group", featureType.getName());