Reprocess operator file settings on service start

docs/changelog/114295.yaml

@@ -0,0 +1,5 @@
+pr: 114295
+summary: "Reprocess operator file settings when settings service starts, due to node restart or master node change"
+area: Infra/Settings
+type: enhancement
+issues: [ ]

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/FileSettingsRoleMappingUpgradeIT.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.upgrades;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.test.XContentTestUtils;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.util.Version;
+import org.elasticsearch.test.cluster.util.resource.Resource;
+import org.elasticsearch.test.junit.RunnableTestRuleAdapter;
+import org.junit.ClassRule;
+import org.junit.rules.RuleChain;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestRule;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+
+public class FileSettingsRoleMappingUpgradeIT extends ParameterizedRollingUpgradeTestCase {
+
+    private static final RunnableTestRuleAdapter versionLimit = new RunnableTestRuleAdapter(
+        () -> assumeTrue(
+            "Only relevant when upgrading from a version before role mappings were stored in cluster state",
+            getOldClusterTestVersion().after(new Version(8, 4, 0)) && getOldClusterTestVersion().before(new Version(8, 15, 0))
+        )
+    );
+
+    private static final String settingsJSON = """
+        {
+             "metadata": {
+                 "version": "1",
+                 "compatibility": "8.4.0"
+             },
+             "state": {
+                 "role_mappings": {
+                       "everyone_kibana": {
+                          "enabled": true,
+                          "roles": [ "kibana_user" ],
+                          "rules": { "field": { "username": "*" } }
+                       }
+                 }
+             }
+        }""";
+
+    private static final TemporaryFolder repoDirectory = new TemporaryFolder();
+
+    private static final ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .distribution(DistributionType.DEFAULT)
+        .version(getOldClusterTestVersion())
+        .nodes(NODE_NUM)
+        .setting("path.repo", new Supplier<>() {
+            @Override
+            @SuppressForbidden(reason = "TemporaryFolder only has io.File methods, not nio.File")
+            public String get() {
+                return repoDirectory.getRoot().getPath();
+            }
+        })
+        .setting("xpack.security.enabled", "true")
+        // workaround to avoid having to set up clients and authorization headers
+        .setting("xpack.security.authc.anonymous.roles", "superuser")
+        .configFile("operator/settings.json", Resource.fromString(settingsJSON))
+        .build();
+
+    @ClassRule
+    public static TestRule ruleChain = RuleChain.outerRule(versionLimit).around(repoDirectory).around(cluster);
+
+    public FileSettingsRoleMappingUpgradeIT(@Name("upgradedNodes") int upgradedNodes) {
+        super(upgradedNodes);
+    }
+
+    @Override
+    protected ElasticsearchCluster getUpgradeCluster() {
+        return cluster;
+    }
+
+    public void testRoleMappingsAppliedOnUpgrade() throws IOException {
+        if (isOldCluster()) {
+            Request clusterStateRequest = new Request("GET", "/_cluster/state/metadata");
+            List<Object> roleMappings = new XContentTestUtils.JsonMapView(entityAsMap(client().performRequest(clusterStateRequest))).get(
+                "metadata.role_mappings.role_mappings"
+            );
+            assertThat(roleMappings, is(nullValue()));
+        } else if (isUpgradedCluster()) {
+            // the nodes have all been upgraded. Check they re-processed the role mappings in the settings file on
+            // upgrade
+            Request clusterStateRequest = new Request("GET", "/_cluster/state/metadata");
+            List<Object> roleMappings = new XContentTestUtils.JsonMapView(entityAsMap(client().performRequest(clusterStateRequest))).get(
+                "metadata.role_mappings.role_mappings"
+            );
+            assertThat(roleMappings, is(not(nullValue())));
+            assertThat(roleMappings.size(), equalTo(1));
+        }
+    }
+}

server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/FileSettingsServiceIT.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.reservedstate.action.ReservedClusterSettingsAction;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.junit.Before;
 
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -40,6 +41,7 @@
 import static org.elasticsearch.test.NodeRoles.dataOnlyNode;
 import static org.elasticsearch.test.NodeRoles.masterNode;
 import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -50,7 +52,12 @@
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, autoManageMasterNodes = false)
 public class FileSettingsServiceIT extends ESIntegTestCase {
 
-    private static final AtomicLong versionCounter = new AtomicLong(1);
+    private final AtomicLong versionCounter = new AtomicLong(1);
+
+    @Before
+    public void resetVersionCounter() {
+        versionCounter.set(1);
+    }
 
     private static final String testJSON = """
         {
@@ -102,15 +109,29 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
              }
         }""";
 
+    private static final String testOtherErrorJSON = """
+        {
+             "metadata": {
+                 "version": "%s",
+                 "compatibility": "8.4.0"
+             },
+             "state": {
+                 "bad_cluster_settings": {
+                     "search.allow_expensive_queries": "false"
+                 }
+             }
+        }""";
+
     private void assertMasterNode(Client client, String node) {
         assertThat(
             client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode().getName(),
             equalTo(node)
         );
     }
 
-    public static void writeJSONFile(String node, String json, AtomicLong versionCounter, Logger logger) throws Exception {
-        long version = versionCounter.incrementAndGet();
+    public static void writeJSONFile(String node, String json, AtomicLong versionCounter, Logger logger, boolean incrementVersion)
+        throws Exception {
+        long version = incrementVersion ? versionCounter.incrementAndGet() : versionCounter.get();
 
         FileSettingsService fileSettingsService = internalCluster().getInstance(FileSettingsService.class, node);
 
@@ -124,6 +145,15 @@ public static void writeJSONFile(String node, String json, AtomicLong versionCou
         logger.info("--> After writing new settings file: [{}]", settingsFileContent);
     }
 
+    public static void writeJSONFile(String node, String json, AtomicLong versionCounter, Logger logger) throws Exception {
+        writeJSONFile(node, json, versionCounter, logger, true);
+    }
+
+    public static void writeJSONFileWithoutVersionIncrement(String node, String json, AtomicLong versionCounter, Logger logger)
+        throws Exception {
+        writeJSONFile(node, json, versionCounter, logger, false);
+    }
+
     private Tuple<CountDownLatch, AtomicLong> setupCleanupClusterStateListener(String node) {
         ClusterService clusterService = internalCluster().clusterService(node);
         CountDownLatch savedClusterState = new CountDownLatch(1);
@@ -171,7 +201,10 @@ public void clusterChanged(ClusterChangedEvent event) {
     private void assertClusterStateSaveOK(CountDownLatch savedClusterState, AtomicLong metadataVersion, String expectedBytesPerSec)
         throws Exception {
         assertTrue(savedClusterState.await(20, TimeUnit.SECONDS));
+        assertExpectedRecoveryBytesSettingAndVersion(metadataVersion, expectedBytesPerSec);
+    }
 
+    private static void assertExpectedRecoveryBytesSettingAndVersion(AtomicLong metadataVersion, String expectedBytesPerSec) {
         final ClusterStateResponse clusterStateResponse = clusterAdmin().state(
             new ClusterStateRequest(TEST_REQUEST_TIMEOUT).waitForMetadataVersion(metadataVersion.get())
         ).actionGet();
@@ -337,6 +370,77 @@ public void testErrorSaved() throws Exception {
         assertClusterStateNotSaved(savedClusterState.v1(), savedClusterState.v2());
     }
 
+    public void testErrorCanRecoverOnRestart() throws Exception {
+        internalCluster().setBootstrapMasterNodeIndex(0);
+        logger.info("--> start data node / non master node");
+        String dataNode = internalCluster().startNode(Settings.builder().put(dataOnlyNode()).put("discovery.initial_state_timeout", "1s"));
+        FileSettingsService dataFileSettingsService = internalCluster().getInstance(FileSettingsService.class, dataNode);
+
+        assertFalse(dataFileSettingsService.watching());
+
+        logger.info("--> start master node");
+        final String masterNode = internalCluster().startMasterOnlyNode(
+            Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
+        );
+        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        var savedClusterState = setupClusterStateListenerForError(masterNode);
+
+        FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
+
+        assertTrue(masterFileSettingsService.watching());
+        assertFalse(dataFileSettingsService.watching());
+
+        writeJSONFile(masterNode, testErrorJSON, versionCounter, logger);
+        AtomicLong metadataVersion = savedClusterState.v2();
+        assertClusterStateNotSaved(savedClusterState.v1(), metadataVersion);
+        assertHasErrors(metadataVersion, "not_cluster_settings");
+
+        // write valid json without version increment to simulate ES being able to process settings after a restart (usually, this would be
+        // due to a code change)
+        writeJSONFileWithoutVersionIncrement(masterNode, testJSON, versionCounter, logger);
+        internalCluster().restartNode(masterNode);
+        ensureGreen();
+
+        // we don't know the exact metadata version to wait for so rely on an assertBusy instead
+        assertBusy(() -> assertExpectedRecoveryBytesSettingAndVersion(metadataVersion, "50mb"));
+        assertBusy(() -> assertNoErrors(metadataVersion));
+    }
+
+    public void testNewErrorOnRestartReprocessing() throws Exception {
+        internalCluster().setBootstrapMasterNodeIndex(0);
+        logger.info("--> start data node / non master node");
+        String dataNode = internalCluster().startNode(Settings.builder().put(dataOnlyNode()).put("discovery.initial_state_timeout", "1s"));
+        FileSettingsService dataFileSettingsService = internalCluster().getInstance(FileSettingsService.class, dataNode);
+
+        assertFalse(dataFileSettingsService.watching());
+
+        logger.info("--> start master node");
+        final String masterNode = internalCluster().startMasterOnlyNode(
+            Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
+        );
+        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        var savedClusterState = setupClusterStateListenerForError(masterNode);
+
+        FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
+
+        assertTrue(masterFileSettingsService.watching());
+        assertFalse(dataFileSettingsService.watching());
+
+        writeJSONFile(masterNode, testErrorJSON, versionCounter, logger);
+        AtomicLong metadataVersion = savedClusterState.v2();
+        assertClusterStateNotSaved(savedClusterState.v1(), metadataVersion);
+        assertHasErrors(metadataVersion, "not_cluster_settings");
+
+        // write json with new error without version increment to simulate ES failing to process settings after a restart for a new reason
+        // (usually, this would be due to a code change)
+        writeJSONFileWithoutVersionIncrement(masterNode, testOtherErrorJSON, versionCounter, logger);
+        assertHasErrors(metadataVersion, "not_cluster_settings");
+        internalCluster().restartNode(masterNode);
+        ensureGreen();
+
+        assertBusy(() -> assertHasErrors(metadataVersion, "bad_cluster_settings"));
+    }
+
     public void testSettingsAppliedOnMasterReElection() throws Exception {
         internalCluster().setBootstrapMasterNodeIndex(0);
         logger.info("--> start master node");
@@ -383,4 +487,21 @@ public void testSettingsAppliedOnMasterReElection() throws Exception {
         assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2(), "43mb");
     }
 
+    private void assertHasErrors(AtomicLong waitForMetadataVersion, String expectedError) {
+        var errorMetadata = getErrorMetadata(waitForMetadataVersion);
+        assertThat(errorMetadata, is(notNullValue()));
+        assertThat(errorMetadata.errors(), containsInAnyOrder(containsString(expectedError)));
+    }
+
+    private void assertNoErrors(AtomicLong waitForMetadataVersion) {
+        var errorMetadata = getErrorMetadata(waitForMetadataVersion);
+        assertThat(errorMetadata, is(nullValue()));
+    }
+
+    private ReservedStateErrorMetadata getErrorMetadata(AtomicLong waitForMetadataVersion) {
+        final ClusterStateResponse clusterStateResponse = clusterAdmin().state(
+            new ClusterStateRequest(TEST_REQUEST_TIMEOUT).waitForMetadataVersion(waitForMetadataVersion.get())
+        ).actionGet();
+        return clusterStateResponse.getState().getMetadata().reservedStateMetadata().get(FileSettingsService.NAMESPACE).errorMetadata();
+    }
 }

server/src/main/java/org/elasticsearch/common/file/AbstractFileWatchingService.java

@@ -77,6 +77,15 @@ public AbstractFileWatchingService(Path watchedFile) {
 
     protected abstract void processInitialFileMissing() throws InterruptedException, ExecutionException, IOException;
 
+    /**
+     * Defaults to generic {@link #processFileChanges()} behavior.
+     * An implementation can override this to define different file handling when the file is processed during
+     * initial service start.
+     */
+    protected void processFileOnServiceStart() throws IOException, ExecutionException, InterruptedException {
+        processFileChanges();
+    }
+
     public final void addFileChangedListener(FileChangedListener listener) {
         eventListeners.add(listener);
     }
@@ -174,7 +183,7 @@ protected final void watcherThread() {
 
             if (Files.exists(path)) {
                 logger.debug("found initial operator settings file [{}], applying...", path);
-                processSettingsAndNotifyListeners();
+                processSettingsOnServiceStartAndNotifyListeners();
             } else {
                 processInitialFileMissing();
                 // Notify everyone we don't have any initial file settings
@@ -290,6 +299,17 @@ final WatchKey enableDirectoryWatcher(WatchKey previousKey, Path settingsDir) th
         } while (true);
     }
 
+    void processSettingsOnServiceStartAndNotifyListeners() throws InterruptedException {
+        try {
+            processFileOnServiceStart();
+            for (var listener : eventListeners) {
+                listener.watchedFileChanged();
+            }
+        } catch (IOException | ExecutionException e) {
+            logger.error(() -> "Error processing watched file: " + watchedFile(), e);
+        }
+    }
+
     void processSettingsAndNotifyListeners() throws InterruptedException {
         try {
             processFileChanges();

server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java

@@ -115,20 +115,34 @@ protected boolean shouldRefreshFileState(ClusterState clusterState) {
      */
     @Override
     protected void processFileChanges() throws ExecutionException, InterruptedException, IOException {
-        PlainActionFuture<Void> completion = new PlainActionFuture<>();
         logger.info("processing path [{}] for [{}]", watchedFile(), NAMESPACE);
+        processFileChanges(false);
+    }
+
+    /**
+     * Read settings and pass them to {@link ReservedClusterStateService} for application.
+     * Settings will be reprocessed even if the cluster-state version equals that found in the settings file.
+     */
+    @Override
+    protected void processFileOnServiceStart() throws IOException, ExecutionException, InterruptedException {
+        logger.info("processing path [{}] for [{}] on service start", watchedFile(), NAMESPACE);
+        processFileChanges(true);
+    }
+
+    private void processFileChanges(boolean allowSameVersion) throws IOException, InterruptedException, ExecutionException {
+        PlainActionFuture<Void> completion = new PlainActionFuture<>();
         try (
             var fis = Files.newInputStream(watchedFile());
             var bis = new BufferedInputStream(fis);
             var parser = JSON.xContent().createParser(XContentParserConfiguration.EMPTY, bis)
         ) {
-            stateService.process(NAMESPACE, parser, (e) -> completeProcessing(e, completion));
+            stateService.process(NAMESPACE, parser, allowSameVersion, (e) -> completeProcessing(e, completion));
         }
         completion.get();
     }
 
     @Override
-    protected void processInitialFileMissing() throws ExecutionException, InterruptedException, IOException {
+    protected void processInitialFileMissing() throws ExecutionException, InterruptedException {
         PlainActionFuture<ActionResponse.Empty> completion = new PlainActionFuture<>();
         logger.info("setting file [{}] not found, initializing [{}] as empty", watchedFile(), NAMESPACE);
         stateService.initEmpty(NAMESPACE, completion);