System index descriptors support mixed versions (#71144)

System index descriptors are used to describe a system index, which are
expected to change as new versions are developed. As part of this, the
descriptors had a minimum supported version field so that the contents
within that descriptor would not be applied if there were nodes older
than that version. However, this falls short of being able to
accurately describe what a system index should look like in a given
cluster where there are mixed node versions.

This change moves us towards being able to accurately describe and
know what the system index should look like. A system index is now
able to accept a list of the prior system index descriptor objects
so that clusters with mixed versions can select the appropriate
descriptor and ensure the index is created properly. As the node
versions change during a rolling upgrade, the cluster will then be
able to adapt the system index to the most recent version once all
master and data nodes have been upgraded.

Co-authored-by: Tim Vernum <tim@adjective.org>
Co-authored-by: Yang Wang <ywangd@gmail.com>

qa/smoke-test-http/src/test/java/org/elasticsearch/http/SystemIndexRestIT.java

@@ -166,7 +166,7 @@ public class SystemIndexRestIT extends HttpSmokeTestCase {
                     .setVersionMetaKey("version")
                     .setMappings(builder)
                     .setSettings(SETTINGS)
-                    .setType(Type.INTERNAL)
+                    .setType(Type.INTERNAL_MANAGED)
                     .build()
                 );
             } catch (IOException e) {

server/src/internalClusterTest/java/org/elasticsearch/indices/TestSystemIndexDescriptor.java

@@ -39,8 +39,8 @@ public class TestSystemIndexDescriptor extends SystemIndexDescriptor {
         .build();
 
     TestSystemIndexDescriptor() {
-        super(INDEX_NAME + "*", PRIMARY_INDEX_NAME, "Test system index", null, SETTINGS, INDEX_NAME, 0, "version", "stack", null,
-            Type.INTERNAL, List.of());
+        super(INDEX_NAME + "*", PRIMARY_INDEX_NAME, "Test system index", getOldMappings(), SETTINGS, INDEX_NAME, 0, "version", "stack",
+            Version.CURRENT.minimumCompatibilityVersion(), Type.INTERNAL_MANAGED, List.of(), List.of());
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/indices/create/AutoCreateAction.java

@@ -141,17 +141,16 @@ public final class AutoCreateAction extends ActionType<CreateIndexResponse> {
                             return currentState;
                         }
 
-                        final SystemIndexDescriptor descriptor = systemIndices.findMatchingDescriptor(indexName);
-                        final boolean isSystemIndex = descriptor != null && descriptor.isAutomaticallyManaged();
+                        final SystemIndexDescriptor mainDescriptor = systemIndices.findMatchingDescriptor(indexName);
+                        final boolean isSystemIndex = mainDescriptor != null && mainDescriptor.isAutomaticallyManaged();
 
                         final CreateIndexClusterStateUpdateRequest updateRequest;
 
                         if (isSystemIndex) {
-                            final String message = descriptor.checkMinimumNodeVersion(
-                                "auto-create index",
-                                state.nodes().getMinNodeVersion()
-                            );
-                            if (message != null) {
+                            final SystemIndexDescriptor descriptor =
+                                mainDescriptor.getDescriptorCompatibleWith(state.nodes().getSmallestNonClientNodeVersion());
+                            if (descriptor == null) {
+                                final String message = mainDescriptor.getMinimumNodeVersionMessage("auto-create index");
                                 logger.warn(message);
                                 throw new IllegalStateException(message);
                             }

server/src/main/java/org/elasticsearch/action/admin/indices/create/TransportCreateIndexAction.java

@@ -70,9 +70,8 @@ public class TransportCreateIndexAction extends TransportMasterNodeAction<Create
         final long resolvedAt = System.currentTimeMillis();
         final String indexName  = indexNameExpressionResolver.resolveDateMathExpression(request.index(), resolvedAt);
 
-        final SystemIndexDescriptor descriptor = systemIndices.findMatchingDescriptor(indexName);
-
-        final boolean isSystemIndex = descriptor != null && descriptor.isAutomaticallyManaged();
+        final SystemIndexDescriptor mainDescriptor = systemIndices.findMatchingDescriptor(indexName);
+        final boolean isSystemIndex = mainDescriptor != null && mainDescriptor.isAutomaticallyManaged();
 
         final CreateIndexClusterStateUpdateRequest updateRequest;
 
@@ -81,8 +80,10 @@ public class TransportCreateIndexAction extends TransportMasterNodeAction<Create
         // We check this via the request's origin. Eventually, `SystemIndexManager` will reconfigure
         // the index to the latest settings.
         if (isSystemIndex && Strings.isNullOrEmpty(request.origin())) {
-            final String message = descriptor.checkMinimumNodeVersion("create index", state.nodes().getMinNodeVersion());
-            if (message != null) {
+            final SystemIndexDescriptor descriptor =
+                mainDescriptor.getDescriptorCompatibleWith(state.nodes().getSmallestNonClientNodeVersion());
+            if (descriptor == null) {
+                final String message = mainDescriptor.getMinimumNodeVersionMessage("create index");
                 logger.warn(message);
                 listener.onFailure(new IllegalStateException(message));
                 return;

server/src/main/java/org/elasticsearch/indices/SystemIndexDescriptor.java

@@ -19,14 +19,17 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * A system index descriptor describes one or more system indices. It can match a number of indices using
@@ -34,7 +37,7 @@ import java.util.Objects;
  * indices that are managed internally to Elasticsearch, a descriptor can also include information for
  * creating the system index, upgrading its mappings, and creating an alias.
  */
-public class SystemIndexDescriptor {
+public class SystemIndexDescriptor implements Comparable<SystemIndexDescriptor> {
     /** A pattern, either with a wildcard or simple regex. Indices that match one of these patterns are considered system indices. */
     private final String indexPattern;
 
@@ -71,9 +74,12 @@ public class SystemIndexDescriptor {
     /** For internally-managed indices, specifies the origin to use when creating or updating the index */
     private final String origin;
 
-    /** The minimum cluster node version required for this descriptor, or null if there is no restriction */
+    /** The minimum cluster node version required for this descriptor */
     private final Version minimumNodeVersion;
 
+    /** Mapping version from the descriptor */
+    private final Version mappingVersion;
+
     /** Whether there are dynamic fields in this descriptor's mappings */
     private final boolean hasDynamicMappings;
 
@@ -83,6 +89,12 @@ public class SystemIndexDescriptor {
     /** A list of allowed product origins that may access an external system index */
     private final List<String> allowedElasticProductOrigins;
 
+    /**
+     * A list of prior system index descriptors that can be used when one or more data/master nodes is on a version lower than the
+     * minimum supported version for this descriptor
+     */
+    private final List<SystemIndexDescriptor> priorSystemIndexDescriptors;
+
     /**
      * Creates a descriptor for system indices matching the supplied pattern. These indices will not be managed
      * by Elasticsearch internally.
@@ -90,7 +102,8 @@ public class SystemIndexDescriptor {
      * @param description The name of the plugin responsible for this system index.
      */
     public SystemIndexDescriptor(String indexPattern, String description) {
-        this(indexPattern, null, description, null, null, null, 0, null, null, null, Type.INTERNAL, List.of());
+        this(indexPattern, null, description, null, null, null, 0, null, null, Version.CURRENT.minimumCompatibilityVersion(),
+            Type.INTERNAL_UNMANAGED, List.of(), List.of());
     }
 
     /**
@@ -103,7 +116,8 @@ public class SystemIndexDescriptor {
      *                                     indices
      */
     public SystemIndexDescriptor(String indexPattern, String description, Type type, List<String> allowedElasticProductOrigins) {
-        this(indexPattern, null, description, null, null, null, 0, null, null, null, type, allowedElasticProductOrigins);
+        this(indexPattern, null, description, null, null, null, 0, null, null, Version.CURRENT.minimumCompatibilityVersion(), type,
+            allowedElasticProductOrigins, List.of());
     }
 
     /**
@@ -111,6 +125,7 @@ public class SystemIndexDescriptor {
      * by Elasticsearch internally if mappings or settings are provided.
      *
      * @param indexPattern The pattern of index names that this descriptor will be used for. Must start with a '.' character.
+     * @param primaryIndex The primary index name of this descriptor. Used when creating the system index for the first time.
      * @param description The name of the plugin responsible for this system index.
      * @param mappings The mappings to apply to this index when auto-creating, if appropriate
      * @param settings The settings to apply to this index when auto-creating, if appropriate
@@ -119,10 +134,12 @@ public class SystemIndexDescriptor {
      * @param versionMetaKey a mapping key under <code>_meta</code> where a version can be found, which indicates the
     *                       Elasticsearch version when the index was created.
      * @param origin the client origin to use when creating this index.
-     * @param minimumNodeVersion the minimum cluster node version required for this descriptor, or null if there is no restriction
+     * @param minimumNodeVersion the minimum cluster node version required for this descriptor
      * @param type The {@link Type} of system index
      * @param allowedElasticProductOrigins A list of allowed origin values that should be allowed access in the case of external system
      *                                     indices
+     * @param priorSystemIndexDescriptors A list of system index descriptors that describe the same index in a way that is compatible with
+     *                                    older versions of Elasticsearch
      */
     SystemIndexDescriptor(
         String indexPattern,
@@ -136,7 +153,8 @@ public class SystemIndexDescriptor {
         String origin,
         Version minimumNodeVersion,
         Type type,
-        List<String> allowedElasticProductOrigins
+        List<String> allowedElasticProductOrigins,
+        List<SystemIndexDescriptor> priorSystemIndexDescriptors
     ) {
         Objects.requireNonNull(indexPattern, "system index pattern must not be null");
         if (indexPattern.length() < 2) {
@@ -176,12 +194,18 @@ public class SystemIndexDescriptor {
 
         Strings.requireNonEmpty(indexPattern, "indexPattern must be supplied");
 
-        if (mappings != null || settings != null) {
-            Strings.requireNonEmpty(primaryIndex, "Must supply primaryIndex if mappings or settings are defined");
-            Strings.requireNonEmpty(versionMetaKey, "Must supply versionMetaKey if mappings or settings are defined");
-            Strings.requireNonEmpty(origin, "Must supply origin if mappings or settings are defined");
-        }
         Objects.requireNonNull(type, "type must not be null");
+        if (type.isManaged()) {
+            Objects.requireNonNull(settings, "Must supply settings for a managed system index");
+            Strings.requireNonEmpty(mappings, "Must supply mappings for a managed system index");
+            Strings.requireNonEmpty(primaryIndex, "Must supply primaryIndex for a managed system index");
+            Strings.requireNonEmpty(versionMetaKey, "Must supply versionMetaKey for a managed system index");
+            Strings.requireNonEmpty(origin, "Must supply origin for a managed system index");
+            this.mappingVersion = extractVersionFromMappings(mappings, versionMetaKey);;
+        } else {
+            this.mappingVersion = null;
+        }
+
         Objects.requireNonNull(allowedElasticProductOrigins, "allowedProductOrigins must not be null");
         if (type.isInternal() && allowedElasticProductOrigins.isEmpty() == false) {
             throw new IllegalArgumentException("Allowed origins are not valid for internal system indices");
@@ -189,6 +213,40 @@ public class SystemIndexDescriptor {
             throw new IllegalArgumentException("External system indices without allowed products is not a valid combination");
         }
 
+        Objects.requireNonNull(minimumNodeVersion, "minimumNodeVersion must be provided!");
+        Objects.requireNonNull(priorSystemIndexDescriptors, "priorSystemIndexDescriptors must not be null");
+        if (priorSystemIndexDescriptors.isEmpty() == false) {
+            // the rules for prior system index descriptors
+            // 1. No values with the same minimum node version
+            // 2. All prior system index descriptors must have a minimumNodeVersion before this one
+            // 3. Prior system index descriptors may not have other prior system index descriptors
+            //    to avoid multiple branches that need followed
+            // 4. Must have same indexPattern, primaryIndex, and alias
+            Set<Version> versions = new HashSet<>(priorSystemIndexDescriptors.size() + 1);
+            versions.add(minimumNodeVersion);
+            for (SystemIndexDescriptor prior : priorSystemIndexDescriptors) {
+                if (versions.add(prior.minimumNodeVersion) == false) {
+                    throw new IllegalArgumentException(prior + " has the same minimum node version as another descriptor");
+                }
+                if (prior.minimumNodeVersion.after(minimumNodeVersion)) {
+                    throw new IllegalArgumentException(prior + " has minimum node version [" + prior.minimumNodeVersion +
+                        "] which is after [" + minimumNodeVersion + "]");
+                }
+                if (prior.priorSystemIndexDescriptors.isEmpty() == false) {
+                    throw new IllegalArgumentException(prior + " has its own prior descriptors but only a depth of 1 is allowed");
+                }
+                if (prior.indexPattern.equals(indexPattern) == false) {
+                    throw new IllegalArgumentException("index pattern must be the same");
+                }
+                if (prior.primaryIndex.equals(primaryIndex) == false) {
+                    throw new IllegalArgumentException("primary index must be the same");
+                }
+                if (prior.aliasName.equals(aliasName) == false) {
+                    throw new IllegalArgumentException("alias name must be the same");
+                }
+            }
+        }
+
         this.indexPattern = indexPattern;
         this.primaryIndex = primaryIndex;
 
@@ -207,6 +265,16 @@ public class SystemIndexDescriptor {
         this.allowedElasticProductOrigins = allowedElasticProductOrigins;
         this.hasDynamicMappings = this.mappings != null
             && findDynamicMapping(XContentHelper.convertToMap(JsonXContent.jsonXContent, mappings, false));
+
+        final List<SystemIndexDescriptor> sortedPriorSystemIndexDescriptors;
+        if (priorSystemIndexDescriptors.isEmpty() || priorSystemIndexDescriptors.size() == 1) {
+            sortedPriorSystemIndexDescriptors = List.copyOf(priorSystemIndexDescriptors);
+        } else {
+            List<SystemIndexDescriptor> copy = new ArrayList<>(priorSystemIndexDescriptors);
+            Collections.sort(copy);
+            sortedPriorSystemIndexDescriptors = List.copyOf(copy);
+        }
+        this.priorSystemIndexDescriptors = sortedPriorSystemIndexDescriptors;
     }
 
 
@@ -286,9 +354,12 @@ public class SystemIndexDescriptor {
         return this.versionMetaKey;
     }
 
+    public Version getMinimumNodeVersion() {
+        return minimumNodeVersion;
+    }
+
     public boolean isAutomaticallyManaged() {
-        // TODO remove mappings/settings check after all internal indices have been migrated
-        return type.isManaged() && (this.mappings != null || this.settings != null);
+        return type.isManaged();
     }
 
     public String getOrigin() {
@@ -311,25 +382,49 @@ public class SystemIndexDescriptor {
         return allowedElasticProductOrigins;
     }
 
+    public Version getMappingVersion() {
+        if (type.isManaged() == false) {
+            throw new IllegalStateException(toString() + " is not managed so there are no mappings or version");
+        }
+        return mappingVersion;
+    }
+
     /**
-     * Checks that this descriptor can be used within this cluster, by comparing the supplied minimum
-     * node version to this descriptor's minimum version.
+     * Gets a standardized message when the node contains a data or master node whose version is less
+     * than that of the minimum supported version of this descriptor and its prior descriptors.
      *
      * @param cause the action being attempted that triggered the check. Used in the error message.
-     * @param actualMinimumNodeVersion the lower node version in the cluster
-     * @return an error message if the lowest node version is lower that the version in this descriptor,
-     * or <code>null</code> if the supplied version is acceptable or this descriptor has no minimum version.
+     * @return the standardized error message
      */
-    public String checkMinimumNodeVersion(String cause, Version actualMinimumNodeVersion) {
+    public String getMinimumNodeVersionMessage(String cause) {
         Objects.requireNonNull(cause);
-        if (this.minimumNodeVersion != null && this.minimumNodeVersion.after(actualMinimumNodeVersion)) {
-            return String.format(
-                Locale.ROOT,
-                "[%s] failed - system index [%s] requires all cluster nodes to be at least version [%s]",
-                cause,
-                this.getPrimaryIndex(),
-                minimumNodeVersion
-            );
+        final Version actualMinimumVersion = priorSystemIndexDescriptors.isEmpty() ? minimumNodeVersion :
+            priorSystemIndexDescriptors.get(priorSystemIndexDescriptors.size() - 1).minimumNodeVersion;
+        return String.format(
+            Locale.ROOT,
+            "[%s] failed - system index [%s] requires all data and master nodes to be at least version [%s]",
+            cause,
+            this.getPrimaryIndex(),
+            actualMinimumVersion
+        );
+    }
+
+    /**
+     * Finds the descriptor that can be used within this cluster, by comparing the supplied minimum
+     * node version to this descriptor's minimum version and the prior descriptors minimum version.
+     *
+     * @param version the lower node version in the cluster
+     * @return <code>null</code> if the lowest node version is lower than the minimum version in this descriptor,
+     * or the appropriate descriptor if the supplied version is acceptable.
+     */
+    public SystemIndexDescriptor getDescriptorCompatibleWith(Version version) {
+        if (minimumNodeVersion.onOrBefore(version)) {
+            return this;
+        }
+        for (SystemIndexDescriptor prior : priorSystemIndexDescriptors) {
+            if (version.onOrAfter(prior.minimumNodeVersion)) {
+                return prior;
+            }
         }
         return null;
     }
@@ -338,20 +433,28 @@ public class SystemIndexDescriptor {
         return new Builder();
     }
 
+    @Override
+    public int compareTo(SystemIndexDescriptor other) {
+        return minimumNodeVersion.compareTo(other.minimumNodeVersion) * -1;
+    }
+
     /**
-     * The specific type of system index that this descriptor represents. System indices have three defined types, which is used to
-     * control behavior. Elasticsearch itself and plugins have system indices that are necessary for their features;
-     * these system indices are referred to as internal system indices. Internal system indices are always managed indices that
-     * Elasticsearch manages.
+     * The specific type of system index that this descriptor represents. System indices can be one of four defined types; the type is used
+     * to control behavior. Elasticsearch itself and plugins have system indices that are necessary for their features;
+     * these system indices are referred to as internal system indices. System indices can also belong to features outside of Elasticsearch
+     * that may be part of other Elastic stack components. These are external system indices as the intent is for these to be accessed via
+     * normal APIs with a special value.
+     *
+     * Within both internal and external system indices, there are two sub-types. The first are those that are managed by Elasticsearch and
+     * will have mappings/settings changed as the cluster itself is upgraded. The second are those managed by the owning applications code
+     * and for those Elasticsearch will not perform any updates.
      *
-     * System indices can also belong to features outside of Elasticsearch that may be part of other Elastic stack components. These are
-     * external system indices as the intent is for these to be accessed via normal APIs with a special value. Within external system
-     * indices, there are two sub-types. The first are those that are managed by Elasticsearch and will have mappings/settings changed as
-     * the cluster itself is upgraded. The second are those managed by the external application and for those Elasticsearch will not
-     * perform any updates.
+     * Internal system indices are almost always managed indices that Elasticsearch manages, but there are cases where the component of
+     * Elasticsearch will need to manage the system indices itself.
      */
     public enum Type {
-        INTERNAL(false, true),
+        INTERNAL_MANAGED(false, true),
+        INTERNAL_UNMANAGED(false, false),
         EXTERNAL_MANAGED(true, true),
         EXTERNAL_UNMANAGED(true, false);
 
@@ -389,9 +492,10 @@ public class SystemIndexDescriptor {
         private int indexFormat = 0;
         private String versionMetaKey = null;
         private String origin = null;
-        private Version minimumNodeVersion = null;
-        private Type type = Type.INTERNAL;
+        private Version minimumNodeVersion = Version.CURRENT.minimumCompatibilityVersion();
+        private Type type = Type.INTERNAL_MANAGED;
         private List<String> allowedElasticProductOrigins = List.of();
+        private List<SystemIndexDescriptor> priorSystemIndexDescriptors = List.of();
 
         private Builder() {}
 
@@ -460,6 +564,11 @@ public class SystemIndexDescriptor {
             return this;
         }
 
+        public Builder setPriorSystemIndexDescriptors(List<SystemIndexDescriptor> priorSystemIndexDescriptors) {
+            this.priorSystemIndexDescriptors = priorSystemIndexDescriptors;
+            return this;
+        }
+
         /**
          * Builds a {@link SystemIndexDescriptor} using the fields supplied to this builder.
          * @return a populated descriptor.
@@ -478,7 +587,8 @@ public class SystemIndexDescriptor {
                 origin,
                 minimumNodeVersion,
                 type,
-                allowedElasticProductOrigins
+                allowedElasticProductOrigins,
+                priorSystemIndexDescriptors
             );
         }
     }
@@ -546,4 +656,23 @@ public class SystemIndexDescriptor {
 
         return false;
     }
+
+    private static Version extractVersionFromMappings(String mappings, String versionMetaKey) {
+        final Map<String, Object> mappingsMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), mappings, false);
+        final Map<String, Object> doc = (Map<String, Object>) mappingsMap.get("_doc");
+        final Map<String, Object> meta;
+        if (doc == null) {
+            meta = (Map<String, Object>) mappingsMap.get("_meta");
+        } else {
+            meta = (Map<String, Object>) doc.get("_meta");
+        }
+        if (meta == null) {
+            throw new IllegalStateException("mappings do not have _meta field");
+        }
+        final String value = (String) meta.get(versionMetaKey);
+        if (value == null) {
+            throw new IllegalArgumentException("mappings do not have a version in _meta." + versionMetaKey);
+        }
+        return Version.fromString(value);
+    }
 }

server/src/main/java/org/elasticsearch/indices/SystemIndexManager.java

@@ -78,7 +78,7 @@ public class SystemIndexManager implements ClusterStateListener {
             return;
         }
 
-        if (state.nodes().getMaxNodeVersion().after(state.nodes().getMinNodeVersion())) {
+        if (state.nodes().getMaxNodeVersion().after(state.nodes().getSmallestNonClientNodeVersion())) {
             logger.debug("Skipping system indices up-to-date check as cluster has mixed versions");
             return;
         }

server/src/test/java/org/elasticsearch/indices/SystemIndexDescriptorTests.java

@@ -8,10 +8,15 @@
 
 package org.elasticsearch.indices;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.indices.SystemIndexDescriptor.Type;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.indices.SystemIndexDescriptor.findDynamicMapping;
@@ -20,6 +25,8 @@ import static org.hamcrest.Matchers.equalTo;
 
 public class SystemIndexDescriptorTests extends ESTestCase {
 
+    private static final String MAPPINGS = "{ \"_doc\": { \"_meta\": { \"version\": \"7.4.0\" } } }";
+
     /**
      * Tests the various validation rules that are applied when creating a new system index descriptor.
      */
@@ -114,4 +121,130 @@ public class SystemIndexDescriptorTests extends ESTestCase {
 
         assertThat(findDynamicMapping(mappings), equalTo(false));
     }
+
+    public void testPriorSystemIndexDescriptorValidation() {
+        SystemIndexDescriptor prior = priorSystemIndexDescriptorBuilder().build();
+
+        // same minimum node version
+        IllegalArgumentException iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+        assertThat(iae.getMessage(), containsString("same minimum node version"));
+
+        // different min version but prior is after latest!
+        iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setMinimumNodeVersion(Version.fromString("6.8.0"))
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+        assertThat(iae.getMessage(), containsString("has minimum node version [7.0.0] which is after [6.8.0]"));
+
+        // prior has another prior!
+        iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setMinimumNodeVersion(Version.V_7_5_0)
+            .setPriorSystemIndexDescriptors(List.of(
+                SystemIndexDescriptor.builder()
+                    .setIndexPattern(".system*")
+                    .setDescription("system stuff")
+                    .setPrimaryIndex(".system-1")
+                    .setAliasName(".system")
+                    .setType(Type.INTERNAL_MANAGED)
+                    .setSettings(Settings.EMPTY)
+                    .setMappings(MAPPINGS)
+                    .setVersionMetaKey("version")
+                    .setOrigin("system")
+                    .setMinimumNodeVersion(Version.V_7_4_1)
+                    .setPriorSystemIndexDescriptors(List.of(prior))
+                    .build()
+            ))
+            .build());
+        assertThat(iae.getMessage(), containsString("has its own prior descriptors"));
+
+        // different index patterns
+        iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setIndexPattern(".system1*")
+            .setMinimumNodeVersion(Version.V_7_5_0)
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+        assertThat(iae.getMessage(), containsString("index pattern must be the same"));
+
+        // different primary index
+        iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setPrimaryIndex(".system-2")
+            .setMinimumNodeVersion(Version.V_7_5_0)
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+        assertThat(iae.getMessage(), containsString("primary index must be the same"));
+
+        // different alias
+        iae = expectThrows(IllegalArgumentException.class, () -> priorSystemIndexDescriptorBuilder()
+            .setAliasName(".system1")
+            .setMinimumNodeVersion(Version.V_7_5_0)
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+        assertThat(iae.getMessage(), containsString("alias name must be the same"));
+
+        // success!
+        assertNotNull(priorSystemIndexDescriptorBuilder()
+            .setMinimumNodeVersion(Version.V_7_5_0)
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build());
+    }
+
+    public void testGetDescriptorCompatibleWith() {
+        final String mappings = "{ \"_doc\": { \"_meta\": { \"version\": \"7.4.0\" } } }";
+        final SystemIndexDescriptor prior = SystemIndexDescriptor.builder()
+            .setIndexPattern(".system*")
+            .setDescription("system stuff")
+            .setPrimaryIndex(".system-1")
+            .setAliasName(".system")
+            .setType(Type.INTERNAL_MANAGED)
+            .setSettings(Settings.EMPTY)
+            .setMappings(mappings)
+            .setVersionMetaKey("version")
+            .setOrigin("system")
+            .setMinimumNodeVersion(Version.V_7_0_0)
+            .build();
+        final SystemIndexDescriptor descriptor = SystemIndexDescriptor.builder()
+            .setIndexPattern(".system*")
+            .setDescription("system stuff")
+            .setPrimaryIndex(".system-1")
+            .setAliasName(".system")
+            .setType(Type.INTERNAL_MANAGED)
+            .setSettings(Settings.EMPTY)
+            .setMappings(mappings)
+            .setVersionMetaKey("version")
+            .setOrigin("system")
+            .setPriorSystemIndexDescriptors(List.of(prior))
+            .build();
+
+        SystemIndexDescriptor compat = descriptor.getDescriptorCompatibleWith(Version.CURRENT);
+        assertSame(descriptor, compat);
+
+        assertNull(descriptor.getDescriptorCompatibleWith(Version.fromString("6.8.0")));
+
+        compat = descriptor.getDescriptorCompatibleWith(Version.CURRENT.minimumCompatibilityVersion());
+        assertSame(descriptor, compat);
+
+        Version priorToMin = VersionUtils.getPreviousVersion(descriptor.getMinimumNodeVersion());
+        compat = descriptor.getDescriptorCompatibleWith(priorToMin);
+        assertSame(prior, compat);
+
+        compat = descriptor.getDescriptorCompatibleWith(
+            VersionUtils.randomVersionBetween(random(), prior.getMinimumNodeVersion(), priorToMin));
+        assertSame(prior, compat);
+    }
+
+    private SystemIndexDescriptor.Builder priorSystemIndexDescriptorBuilder() {
+        return SystemIndexDescriptor.builder()
+            .setIndexPattern(".system*")
+            .setDescription("system stuff")
+            .setPrimaryIndex(".system-1")
+            .setAliasName(".system")
+            .setType(Type.INTERNAL_MANAGED)
+            .setSettings(Settings.EMPTY)
+            .setMappings(MAPPINGS)
+            .setVersionMetaKey("version")
+            .setOrigin("system")
+            .setMinimumNodeVersion(Version.V_7_0_0);
+    }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/SecurityIndexManager.java

@@ -191,7 +191,7 @@ public class SecurityIndexManager implements ClusterStateListener {
         }
         final String indexUUID = indexMetadata != null ? indexMetadata.getIndexUUID() : null;
         final State newState = new State(creationTime, isIndexUpToDate, indexAvailable, mappingIsUpToDate, mappingVersion,
-                concreteIndexName, indexHealth, indexState, event.state().nodes().getMinNodeVersion(), indexUUID);
+                concreteIndexName, indexHealth, indexState, event.state().nodes().getSmallestNonClientNodeVersion(), indexUUID);
         this.indexState = newState;
 
         if (newState.equals(previousState) == false) {
@@ -222,14 +222,20 @@ public class SecurityIndexManager implements ClusterStateListener {
     }
 
     private boolean checkIndexMappingUpToDate(ClusterState clusterState) {
+        final Version minimumNonClientNodeVersion = clusterState.nodes().getSmallestNonClientNodeVersion();
+        final SystemIndexDescriptor descriptor = systemIndexDescriptor.getDescriptorCompatibleWith(minimumNonClientNodeVersion);
+        if (descriptor == null) {
+            return false;
+        }
+
         /*
          * The method reference looks wrong here, but it's just counter-intuitive. It expands to:
          *
-         *     mappingVersion -> Version.CURRENT.onOrBefore(mappingVersion)
+         *     mappingVersion -> descriptor.getMappingVersion().onOrBefore(mappingVersion)
          *
          * ...which is true if the mappings have been updated.
          */
-        return checkIndexMappingVersionMatches(clusterState, Version.CURRENT::onOrBefore);
+        return checkIndexMappingVersionMatches(clusterState, descriptor.getMappingVersion()::onOrBefore);
     }
 
     private boolean checkIndexMappingVersionMatches(ClusterState clusterState, Predicate<Version> predicate) {
@@ -327,59 +333,69 @@ public class SecurityIndexManager implements ClusterStateListener {
                         + "Security features relying on the index will not be available until the upgrade API is run on the index");
             } else if (indexState.indexExists() == false) {
                 assert indexState.concreteIndexName != null;
-                logger.info(
-                    "security index does not exist, creating [{}] with alias [{}]",
-                    indexState.concreteIndexName,
-                    systemIndexDescriptor.getAliasName()
-                );
-
-                // Although `TransportCreateIndexAction` is capable of automatically applying the right mappings, settings and aliases for
-                // system indices, we nonetheless specify them here so that the values from `this.systemIndexDescriptor` are used.
-                CreateIndexRequest request = new CreateIndexRequest(indexState.concreteIndexName)
-                    .origin(systemIndexDescriptor.getOrigin())
-                    .mapping(systemIndexDescriptor.getMappings())
-                    .settings(systemIndexDescriptor.getSettings())
-                    .alias(new Alias(systemIndexDescriptor.getAliasName()))
-                    .waitForActiveShards(ActiveShardCount.ALL);
-
-                executeAsyncWithOrigin(client.threadPool().getThreadContext(), systemIndexDescriptor.getOrigin(), request,
-                    new ActionListener<CreateIndexResponse>() {
-                        @Override
-                        public void onResponse(CreateIndexResponse createIndexResponse) {
-                            if (createIndexResponse.isAcknowledged()) {
-                                andThen.run();
-                            } else {
-                                consumer.accept(new ElasticsearchException("Failed to create security index"));
+                final SystemIndexDescriptor descriptorForVersion =
+                    systemIndexDescriptor.getDescriptorCompatibleWith(indexState.minimumNodeVersion);
+
+                if (descriptorForVersion == null) {
+                    final String error = systemIndexDescriptor.getMinimumNodeVersionMessage("create index");
+                    consumer.accept(new IllegalStateException(error));
+                } else {
+                    logger.info(
+                        "security index does not exist, creating [{}] with alias [{}]",
+                        indexState.concreteIndexName,
+                        descriptorForVersion.getAliasName()
+                    );
+                    // Although `TransportCreateIndexAction` is capable of automatically applying the right mappings, settings and aliases
+                    // for system indices, we nonetheless specify them here so that the values from `descriptorForVersion` are used.
+                    CreateIndexRequest request = new CreateIndexRequest(indexState.concreteIndexName)
+                        .origin(descriptorForVersion.getOrigin())
+                        .mapping(descriptorForVersion.getMappings())
+                        .settings(descriptorForVersion.getSettings())
+                        .alias(new Alias(descriptorForVersion.getAliasName()))
+                        .waitForActiveShards(ActiveShardCount.ALL);
+
+                    executeAsyncWithOrigin(client.threadPool().getThreadContext(), descriptorForVersion.getOrigin(), request,
+                        new ActionListener<CreateIndexResponse>() {
+                            @Override
+                            public void onResponse(CreateIndexResponse createIndexResponse) {
+                                if (createIndexResponse.isAcknowledged()) {
+                                    andThen.run();
+                                } else {
+                                    consumer.accept(new ElasticsearchException("Failed to create security index"));
+                                }
                             }
-                        }
-
-                        @Override
-                        public void onFailure(Exception e) {
-                            final Throwable cause = ExceptionsHelper.unwrapCause(e);
-                            if (cause instanceof ResourceAlreadyExistsException) {
-                                // the index already exists - it was probably just created so this
-                                // node hasn't yet received the cluster state update with the index
-                                andThen.run();
-                            } else {
-                                consumer.accept(e);
+
+                            @Override
+                            public void onFailure(Exception e) {
+                                final Throwable cause = ExceptionsHelper.unwrapCause(e);
+                                if (cause instanceof ResourceAlreadyExistsException) {
+                                    // the index already exists - it was probably just created so this
+                                    // node hasn't yet received the cluster state update with the index
+                                    andThen.run();
+                                } else {
+                                    consumer.accept(e);
+                                }
                             }
-                        }
-                    }, client.admin().indices()::create);
+                        }, client.admin().indices()::create
+                    );
+                }
             } else if (indexState.mappingUpToDate == false) {
-                final String error = systemIndexDescriptor.checkMinimumNodeVersion("create index", indexState.minimumNodeVersion);
-                if (error != null) {
+                final SystemIndexDescriptor descriptorForVersion =
+                    systemIndexDescriptor.getDescriptorCompatibleWith(indexState.minimumNodeVersion);
+                if (descriptorForVersion == null) {
+                    final String error = systemIndexDescriptor.getMinimumNodeVersionMessage("updating mapping");
                     consumer.accept(new IllegalStateException(error));
                 } else {
                     logger.info(
                         "Index [{}] (alias [{}]) is not up to date. Updating mapping",
                         indexState.concreteIndexName,
-                        systemIndexDescriptor.getAliasName()
+                        descriptorForVersion.getAliasName()
                     );
                     PutMappingRequest request = new PutMappingRequest(indexState.concreteIndexName).source(
-                        systemIndexDescriptor.getMappings(),
+                        descriptorForVersion.getMappings(),
                         XContentType.JSON
-                    ).origin(systemIndexDescriptor.getOrigin());
-                    executeAsyncWithOrigin(client.threadPool().getThreadContext(), systemIndexDescriptor.getOrigin(), request,
+                    ).origin(descriptorForVersion.getOrigin());
+                    executeAsyncWithOrigin(client.threadPool().getThreadContext(), descriptorForVersion.getOrigin(), request,
                         ActionListener.<AcknowledgedResponse>wrap(putMappingResponse -> {
                             if (putMappingResponse.isAcknowledged()) {
                                 andThen.run();

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/support/SecurityIndexManagerTests.java

@@ -65,8 +65,9 @@ import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
-import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
@@ -280,7 +281,8 @@ public class SecurityIndexManagerTests extends ESTestCase {
         final AtomicReference<Exception> prepareException = new AtomicReference<>(null);
 
         // Hard-code a failure here.
-        when(descriptorSpy.checkMinimumNodeVersion(anyString(), any(Version.class))).thenReturn("Nope");
+        doReturn("Nope").when(descriptorSpy).getMinimumNodeVersionMessage(anyString());
+        doReturn(null).when(descriptorSpy).getDescriptorCompatibleWith(eq(Version.CURRENT));
 
         // Ensure that the mappings for the index are out-of-date, so that the security index manager will
         // attempt to update them.