Archive unknown or invalid settings on updates (#28888)

Today we can end up in a situation where the cluster state contains
unknown or invalid settings. This can happen easily during a rolling
upgrade. For example, consider two nodes that are on a version that
considers the setting foo.bar to be known and valid. Assume one of these
nodes is restarted on a higher version that considers foo.bar to now be
either unknown or invalid, and then the second node is restarted
too. Now, both nodes will be on a version that consider foo.bar to be
unknown or invalid yet this setting will still be contained in the
cluster state. This means that if a cluster settings update is applied
and we validate the settings update with the existing settings then
validation will fail. In such a state, the offending setting can not
even be removed. This commit helps out with this situation by archiving
any settings that are unknown or invalid at the time that a settings
update is applied. This allows the setting update to go through, and the
archived settings can be removed at a later time.

Author: jasontedor

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

@@ -19,13 +19,20 @@
 
 package org.elasticsearch.action.admin.cluster.settings;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 
+import java.util.Map;
+
 import static org.elasticsearch.cluster.ClusterState.builder;
+import static org.elasticsearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
 
 /**
  * Updates transient and persistent cluster state settings if there are any changes
@@ -48,15 +55,34 @@ synchronized Settings getPersistentUpdate() {
         return persistentUpdates.build();
     }
 
-    synchronized ClusterState updateSettings(final ClusterState currentState, Settings transientToApply, Settings persistentToApply) {
+    synchronized ClusterState updateSettings(
+            final ClusterState currentState, final Settings transientToApply, final Settings persistentToApply, final Logger logger) {
         boolean changed = false;
-        Settings.Builder transientSettings = Settings.builder();
-        transientSettings.put(currentState.metaData().transientSettings());
-        changed |= clusterSettings.updateDynamicSettings(transientToApply, transientSettings, transientUpdates, "transient");
 
+        /*
+         * Our cluster state could have unknown or invalid settings that are known and valid in a previous version of Elasticsearch. We can
+         * end up in this situation during a rolling upgrade where the previous version will infect the current version of Elasticsearch
+         * with settings that the current version either no longer knows about or now considers to have invalid values. When the current
+         * version of Elasticsearch becomes infected with a cluster state containing such settings, we need to skip validating such settings
+         * and instead archive them. Consequently, for the current transient and persistent settings in the cluster state we do the
+         * following:
+         *  - split existing settings instance into two with the known and valid settings in one, and the unknown or invalid in another
+         *    (note that existing archived settings are included in the known and valid settings)
+         *  - validate the incoming settings update combined with the existing known and valid settings
+         *  - merge in the archived unknown or invalid settings
+         */
+        final Tuple<Settings, Settings> partitionedTransientSettings =
+                partitionKnownAndValidSettings(currentState.metaData().transientSettings(), "transient", logger);
+        final Settings knownAndValidTransientSettings = partitionedTransientSettings.v1();
+        final Settings unknownOrInvalidTransientSettings = partitionedTransientSettings.v2();
+        final Settings.Builder transientSettings = Settings.builder().put(knownAndValidTransientSettings);
+        changed |= clusterSettings.updateDynamicSettings(transientToApply, transientSettings, transientUpdates, "transient");
 
-        Settings.Builder persistentSettings = Settings.builder();
-        persistentSettings.put(currentState.metaData().persistentSettings());
+        final Tuple<Settings, Settings> partitionedPersistentSettings =
+                partitionKnownAndValidSettings(currentState.metaData().persistentSettings(), "persistent", logger);
+        final Settings knownAndValidPersistentSettings = partitionedPersistentSettings.v1();
+        final Settings unknownOrInvalidPersistentSettings = partitionedPersistentSettings.v2();
+        final Settings.Builder persistentSettings = Settings.builder().put(knownAndValidPersistentSettings);
         changed |= clusterSettings.updateDynamicSettings(persistentToApply, persistentSettings, persistentUpdates, "persistent");
 
         final ClusterState clusterState;
@@ -69,8 +95,8 @@ synchronized ClusterState updateSettings(final ClusterState currentState, Settin
             clusterSettings.validate(persistentFinalSettings, true);
 
             MetaData.Builder metaData = MetaData.builder(currentState.metaData())
-                    .persistentSettings(persistentFinalSettings)
-                    .transientSettings(transientFinalSettings);
+                    .transientSettings(Settings.builder().put(transientFinalSettings).put(unknownOrInvalidTransientSettings).build())
+                    .persistentSettings(Settings.builder().put(persistentFinalSettings).put(unknownOrInvalidPersistentSettings).build());
 
             ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
             boolean updatedReadOnly = MetaData.SETTING_READ_ONLY_SETTING.get(metaData.persistentSettings())
@@ -102,5 +128,46 @@ synchronized ClusterState updateSettings(final ClusterState currentState, Settin
         return clusterState;
     }
 
+    /**
+     * Partitions the settings into those that are known and valid versus those that are unknown or invalid. The resulting tuple contains
+     * the known and valid settings in the first component and the unknown or invalid settings in the second component. Note that archived
+     * settings contained in the settings to partition are included in the first component.
+     *
+     * @param settings     the settings to partition
+     * @param settingsType a string to identify the settings (for logging)
+     * @param logger       a logger to sending warnings to
+     * @return the partitioned settings
+     */
+    private Tuple<Settings, Settings> partitionKnownAndValidSettings(
+            final Settings settings, final String settingsType, final Logger logger) {
+        final Settings existingArchivedSettings = settings.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX));
+        final Settings settingsExcludingExistingArchivedSettings =
+                settings.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX) == false);
+        final Settings settingsWithUnknownOrInvalidArchived = clusterSettings.archiveUnknownOrInvalidSettings(
+                settingsExcludingExistingArchivedSettings,
+                e -> logUnknownSetting(settingsType, e, logger),
+                (e, ex) -> logInvalidSetting(settingsType, e, ex, logger));
+        return Tuple.tuple(
+                Settings.builder()
+                        .put(settingsWithUnknownOrInvalidArchived.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX) == false))
+                        .put(existingArchivedSettings)
+                        .build(),
+                settingsWithUnknownOrInvalidArchived.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX)));
+    }
+
+    private void logUnknownSetting(final String settingType, final Map.Entry<String, String> e, final Logger logger) {
+        logger.warn("ignoring existing unknown {} setting: [{}] with value [{}]; archiving", settingType, e.getKey(), e.getValue());
+    }
+
+    private void logInvalidSetting(
+            final String settingType, final Map.Entry<String, String> e, final IllegalArgumentException ex, final Logger logger) {
+        logger.warn(
+                (Supplier<?>)
+                        () -> new ParameterizedMessage("ignoring existing invalid {} setting: [{}] with value [{}]; archiving",
+                                settingType,
+                                e.getKey(),
+                                e.getValue()),
+                ex);
+    }
 
 }

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

@@ -180,7 +180,8 @@ public void onFailure(String source, Exception e) {
 
             @Override
             public ClusterState execute(final ClusterState currentState) {
-                ClusterState clusterState = updater.updateSettings(currentState, request.transientSettings(), request.persistentSettings());
+                ClusterState clusterState =
+                        updater.updateSettings(currentState, request.transientSettings(), request.persistentSettings(), logger);
                 changed = clusterState != currentState;
                 return clusterState;
             }