[WIP] Operator and plugin read-only settings

server/src/internalClusterTest/java/org/elasticsearch/threadpool/SimpleThreadPoolIT.java

@@ -81,8 +81,10 @@ public void testThreadNames() throws Exception {
             // or the ones that are occasionally come up from ESSingleNodeTestCase
             if (threadName.contains("[node_s_0]") // TODO: this can't possibly be right! single node and integ test are unrelated!
                 || threadName.contains("Keep-Alive-Timer")
+                || threadName.contains("readiness-service")
                 || threadName.contains("JVMCI-native") // GraalVM Compiler Thread
-                || threadName.contains("readiness-service")) {
+                || threadName.contains("file-settings-watcher")
+                || threadName.contains("FileSystemWatchService")) {
                 continue;
             }
             String nodePrefix = "("

server/src/main/java/module-info.java

@@ -272,6 +272,7 @@
     exports org.elasticsearch.monitor.os;
     exports org.elasticsearch.monitor.process;
     exports org.elasticsearch.node;
+    exports org.elasticsearch.operator;
     exports org.elasticsearch.persistent;
     exports org.elasticsearch.persistent.decider;
     exports org.elasticsearch.plugins;

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -249,6 +249,7 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.NamedRegistry;
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.TypeLiteral;
@@ -265,12 +266,17 @@
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.indices.store.TransportNodesListShardStoreMetadata;
+import org.elasticsearch.operator.OperatorHandler;
+import org.elasticsearch.operator.OperatorHandlerProvider;
+import org.elasticsearch.operator.action.OperatorClusterUpdateSettingsAction;
+import org.elasticsearch.operator.service.OperatorClusterStateController;
 import org.elasticsearch.persistent.CompletionPersistentTaskAction;
 import org.elasticsearch.persistent.RemovePersistentTaskAction;
 import org.elasticsearch.persistent.StartPersistentTaskAction;
 import org.elasticsearch.persistent.UpdatePersistentTaskStatusAction;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.ActionPlugin.ActionHandler;
+import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.plugins.interceptor.RestInterceptorActionPlugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
@@ -445,6 +451,8 @@ public class ActionModule extends AbstractModule {
     private final RequestValidators<PutMappingRequest> mappingRequestValidators;
     private final RequestValidators<IndicesAliasesRequest> indicesAliasesRequestRequestValidators;
     private final ThreadPool threadPool;
+    private final OperatorClusterStateController operatorController;
+    private final ClusterService clusterService;
 
     public ActionModule(
         Settings settings,
@@ -457,7 +465,8 @@ public ActionModule(
         NodeClient nodeClient,
         CircuitBreakerService circuitBreakerService,
         UsageService usageService,
-        SystemIndices systemIndices
+        SystemIndices systemIndices,
+        ClusterService clusterService
     ) {
         this.settings = settings;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
@@ -466,6 +475,7 @@ public ActionModule(
         this.settingsFilter = settingsFilter;
         this.actionPlugins = actionPlugins;
         this.threadPool = threadPool;
+        this.clusterService = clusterService;
         actions = setupActions(actionPlugins);
         actionFilters = setupActionFilters(actionPlugins);
         autoCreateIndex = new AutoCreateIndex(settings, clusterSettings, indexNameExpressionResolver, systemIndices);
@@ -508,6 +518,7 @@ public ActionModule(
         );
 
         restController = new RestController(headers, restInterceptor, nodeClient, circuitBreakerService, usageService);
+        operatorController = new OperatorClusterStateController(clusterService);
     }
 
     public Map<String, ActionHandler<?, ?>> getActions() {
@@ -882,6 +893,22 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestCatAction(catActions));
     }
 
+    /**
+     * Initializes the operator action handlers for Elasticsearch and it's modules/plugins
+     *
+     * @param pluginsService needed to load all modules/plugins operator handlers through SPI
+     */
+    public void initOperatorHandlers(PluginsService pluginsService) {
+        List<OperatorHandler<?>> handlers = new ArrayList<>();
+
+        List<? extends OperatorHandlerProvider> pluginHandlers = pluginsService.loadServiceProviders(OperatorHandlerProvider.class);
+
+        handlers.add(new OperatorClusterUpdateSettingsAction(clusterSettings));
+        pluginHandlers.forEach(h -> handlers.addAll(h.handlers()));
+
+        operatorController.initHandlers(handlers);
+    }
+
     @Override
     protected void configure() {
         bind(ActionFilters.class).toInstance(actionFilters);
@@ -914,4 +941,8 @@ public ActionFilters getActionFilters() {
     public RestController getRestController() {
         return restController;
     }
+
+    public OperatorClusterStateController getOperatorController() {
+        return operatorController;
+    }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java

@@ -29,11 +29,13 @@
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.operator.action.OperatorClusterUpdateSettingsAction;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Set;
 
 import static org.elasticsearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
@@ -128,6 +130,17 @@ private static boolean checkClearedBlockAndArchivedSettings(
         return true;
     }
 
+    @Override
+    protected Optional<String> operatorHandlerName() {
+        return Optional.of(OperatorClusterUpdateSettingsAction.NAME);
+    }
+
+    @Override
+    protected Set<String> modifiedKeys(ClusterUpdateSettingsRequest request) {
+        Settings allSettings = Settings.builder().put(request.persistentSettings()).put(request.transientSettings()).build();
+        return allSettings.keySet();
+    }
+
     private static final String UPDATE_TASK_SOURCE = "cluster_update_settings";
     private static final String REROUTE_TASK_SOURCE = "reroute_after_cluster_update_settings";
 
@@ -243,6 +256,10 @@ public ClusterUpdateSettingsTask(
             this.request = request;
         }
 
+        public ClusterUpdateSettingsTask(final ClusterSettings clusterSettings, ClusterUpdateSettingsRequest request) {
+            this(clusterSettings, Priority.IMMEDIATE, request, null);
+        }
+
         @Override
         public ClusterState execute(final ClusterState currentState) {
             final ClusterState clusterState = updater.updateSettings(

server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.OperatorMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -42,6 +43,11 @@
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
 import java.util.function.Predicate;
 
 import static org.elasticsearch.core.Strings.format;
@@ -142,9 +148,62 @@ private ClusterBlockException checkBlockIfStateRecovered(Request request, Cluste
         }
     }
 
+    /**
+     * Override this method if the master node action also has an {@link org.elasticsearch.operator.OperatorHandler}
+     * interaction. We need to check if certain settings or entities are allowed to be modified by the master node
+     * action, depending on if they are set already in operator mode.
+     *
+     * @return an Optional of the operator handler name
+     */
+    protected Optional<String> operatorHandlerName() {
+        return Optional.empty();
+    }
+
+    /**
+     * Override this method to return the keys of the cluster state or cluster entities that are modified by
+     * the Request object. This method is used by the operator handler logic (see {@link org.elasticsearch.operator.OperatorHandler})
+     * to verify if the keys don't conflict with an existing key set in operator mode.
+     *
+     * @param request the TransportMasterNode request
+     * @return set of String keys intended to be modified/set/deleted by this request
+     */
+    protected Set<String> modifiedKeys(Request request) {
+        return Collections.emptySet();
+    }
+
+    // package private for testing
+    void validateForOperatorState(Request request, ClusterState state) {
+        Optional<String> handlerName = operatorHandlerName();
+        assert handlerName.isPresent();
+
+        Set<String> modified = modifiedKeys(request);
+        List<String> errors = new ArrayList<>();
+
+        for (OperatorMetadata operator : state.metadata().operatorState().values()) {
+            Set<String> conflicts = operator.conflicts(handlerName.get(), modified);
+            if (conflicts.isEmpty() == false) {
+                errors.add(format("[%s] set in operator mode by [%s]", String.join(",", conflicts), operator.namespace()));
+            }
+        }
+
+        if (errors.isEmpty() == false) {
+            throw new IllegalArgumentException(
+                format("Failed to process request [%s] with errors: %s", request, String.join(System.lineSeparator(), errors))
+            );
+        }
+    }
+
+    // package private for testing
+    boolean supportsOperatorMode() {
+        return operatorHandlerName().isPresent();
+    }
+
     @Override
     protected void doExecute(Task task, final Request request, ActionListener<Response> listener) {
         ClusterState state = clusterService.state();
+        if (supportsOperatorMode()) {
+            validateForOperatorState(request, state);
+        }
         logger.trace("starting processing request [{}] with cluster state version [{}]", request, state.version());
         if (task != null) {
             request.setParentTask(clusterService.localNode().getId(), task.getId());