[8.x] Reprocess operator file settings on service start (#114295) (#115198)

* Reprocess operator file settings on service start (#114295)

Changes `FileSettingsService` to reprocess file settings on every
restart or master node change, even if versions match between file and
cluster-state metadata. If the file version is lower than the metadata
version, processing is still skipped to avoid applying stale settings. 

This makes it easier for consumers of file settings to change their
behavior w.r.t. file settings contents. For instance, an update of how
role mappings are stored will automatically apply on the next restart,
without the need to manually increment the file settings version to
force reprocessing. 

Relates: ES-9628

* Backport 114295

---------

Co-authored-by: Elastic Machine <[email protected]>

Author: n1v0lg

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.resource.Resource;
+import org.junit.Before;
+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 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(repoDirectory).around(cluster);
+
+    public FileSettingsRoleMappingUpgradeIT(@Name("upgradedNodes") int upgradedNodes) {
+        super(upgradedNodes);
+    }
+
+    @Override
+    protected ElasticsearchCluster getUpgradeCluster() {
+        return cluster;
+    }
+
+    @Before
+    public void checkVersions() {
+        assumeTrue(
+            "Only relevant when upgrading from a version before role mappings were stored in cluster state",
+            oldClusterHasFeature("gte_v8.4.0") && oldClusterHasFeature("gte_v8.15.0") == false
+        );
+    }
+
+    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/ErrorState.java

@@ -15,9 +15,22 @@
 
 import static org.elasticsearch.ExceptionsHelper.stackTrace;
 
-record ErrorState(String namespace, Long version, List<String> errors, ReservedStateErrorMetadata.ErrorKind errorKind) {
-    ErrorState(String namespace, Long version, Exception e, ReservedStateErrorMetadata.ErrorKind errorKind) {
-        this(namespace, version, List.of(stackTrace(e)), errorKind);
+record ErrorState(
+    String namespace,
+    Long version,
+    ReservedStateVersionCheck versionCheck,
+    List<String> errors,
+    ReservedStateErrorMetadata.ErrorKind errorKind
+) {
+
+    ErrorState(
+        String namespace,
+        Long version,
+        ReservedStateVersionCheck versionCheck,
+        Exception e,
+        ReservedStateErrorMetadata.ErrorKind errorKind
+    ) {
+        this(namespace, version, versionCheck, List.of(stackTrace(e)), errorKind);
     }
 
     public String toString() {