Protect replicated data streams against local rollovers (#64710)

When a data stream is being auto followed then a rollover in a local cluster can break auto following,
if the local cluster performs a rollover then it creates a new write index and if then later the remote
cluster rolls over as well then that new write index can't be replicated, because it has the same name
as in the write index in the local cluster, which was created earlier.

If a data stream is managed by ccr, then the local cluster should not do a rollover for those data streams.
The data stream should be rolled over in the remote cluster and that change should replicate to the local
cluster. Performing a rollover in the local cluster is an operation that the data stream support in ccr should
perform.

To protect against rolling over a replicated data stream, this PR adds a replicate field to DataStream class.
The rollover api will fail with an error in case a data stream is being rolled over and the targeted data stream is
a replicated data stream. When the put follow api creates a data stream in the local cluster then the replicate flag
is set to true. There should be a way to turn a replicated data stream into a regular data stream when for example
during disaster recovery. The newly added api in this pr (promote data stream api) is doing that. After a replicated
data stream is promoted to a regular data stream then the local data stream can be rolled over, so that the new
write index is no longer a follower index. Also if the put follow api is attempting to update this data stream
(for example to attempt to resume auto following) then that with fail, because the data stream is no longer a
replicated data stream.

Today with time based indices behind an alias, the is_write_index property isn't replicated from remote cluster
to the local cluster, so when attempting to rollover the alias in the local cluster the rollover fails, because the
alias doesn't have a write index. The added replicated field in the DataStream class and added validation
achieve the same kind of protection, but in a more robust way.

A followup from #61993.

Author: martijnvg

docs/reference/ccr/auto-follow.asciidoc

@@ -13,6 +13,9 @@ automatically followed if the data stream name matches an auto-follow
 pattern. If you create a data stream after creating the auto-follow pattern,
 all backing indices are followed automatically.
 
+The data streams replicated from a remote cluster by CCR are protected from
+local rollovers. The <<promote-data-stream-api,promote data stream API>>
+can be used to turn these data streams into regular data streams.
 
 Auto-follow patterns are especially useful with
 <<index-lifecycle-management,{ilm-cap}>>, which might continually create

docs/reference/data-streams/data-stream-apis.asciidoc

@@ -9,6 +9,7 @@ The following APIs are available for managing <<data-streams,data streams>>:
 * <<indices-get-data-stream>>
 * <<indices-migrate-to-data-stream>>
 * <<data-stream-stats-api>>
+* <<promote-data-stream-api>>
 
 For concepts and tutorials, see <<data-streams>>.
 
@@ -21,3 +22,5 @@ include::{es-repo-dir}/indices/get-data-stream.asciidoc[]
 include::{es-repo-dir}/indices/migrate-to-data-stream.asciidoc[]
 
 include::{es-repo-dir}/indices/data-stream-stats.asciidoc[]
+
+include::{es-repo-dir}/data-streams/promote-data-stream-api.asciidoc[]

docs/reference/data-streams/promote-data-stream-api.asciidoc

@@ -0,0 +1,38 @@
+[role="xpack"]
+[[promote-data-stream-api]]
+=== Promote Data Stream API
+++++
+<titleabbrev>Promote data stream api</titleabbrev>
+++++
+
+The purpose of the promote data stream api is to turn
+a data stream that is replicated by CCR into a regular
+data stream.
+
+Via CCR Auto Following, a data stream from a remote cluster
+can  be replicated to the local cluster. These data streams
+can't be rolled over in the local cluster. Only if the upstream
+data stream rolls over then these replicated data streams roll
+over as well. In the event that the remote cluster is no longer
+available, the data stream in the local cluster can be promoted
+to a regular data stream, which allows these data streams to
+be rolled over in the local cluster.
+
+[source,console]
+----
+POST /_data_stream/_promote/my-data-stream
+----
+// TEST[catch:missing]
+
+[[promote-data-stream-api-request]]
+==== {api-request-title}
+
+`POST /_data_stream/_promote/<data-stream>`
+
+
+[[promote-data-stream-api-path-params]]
+==== {api-path-parms-title}
+
+`<data-stream>`::
+(Required, string)
+The name of the data stream to promote.

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/MetadataRolloverService.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.List;
@@ -148,6 +149,7 @@ private RolloverResult rolloverDataStream(ClusterState currentState, IndexAbstra
         final DataStream ds = dataStream.getDataStream();
         final IndexMetadata originalWriteIndex = dataStream.getWriteIndex();
         final String newWriteIndexName = DataStream.getDefaultBackingIndexName(ds.getName(), ds.getGeneration() + 1);
+        ds.rollover(new Index(newWriteIndexName, "uuid")); // just for validation
         createIndexService.validateIndexName(newWriteIndexName, currentState); // fails if the index already exists
         if (onlyValidate) {
             return new RolloverResult(newWriteIndexName, originalWriteIndex.getIndex().getName(), currentState);

server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java

@@ -44,26 +44,29 @@ public final class DataStream extends AbstractDiffable<DataStream> implements To
 
     public static final String BACKING_INDEX_PREFIX = ".ds-";
     public static final Version HIDDEN_VERSION = Version.V_7_11_0;
+    public static final Version REPLICATED_VERSION = Version.V_8_0_0;
 
     private final String name;
     private final TimestampField timeStampField;
     private final List<Index> indices;
     private final long generation;
     private final Map<String, Object> metadata;
     private final boolean hidden;
+    private final boolean replicated;
 
     public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata) {
-        this(name, timeStampField, indices, generation, metadata, false);
+        this(name, timeStampField, indices, generation, metadata, false, false);
     }
 
     public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata,
-                      boolean hidden) {
+                      boolean hidden, boolean replicated) {
         this.name = name;
         this.timeStampField = timeStampField;
         this.indices = Collections.unmodifiableList(indices);
         this.generation = generation;
         this.metadata = metadata;
         this.hidden = hidden;
+        this.replicated = replicated;
         assert indices.size() > 0;
     }
 
@@ -100,6 +103,16 @@ public boolean isHidden() {
         return hidden;
     }
 
+    /**
+     * Determines whether this data stream is replicated from elsewhere,
+     * for example a remote cluster.
+     *
+     * @return Whether this data stream is replicated.
+     */
+    public boolean isReplicated() {
+        return replicated;
+    }
+
     /**
      * Performs a rollover on a {@code DataStream} instance and returns a new instance containing
      * the updated list of backing indices and incremented generation.
@@ -110,9 +123,14 @@ public boolean isHidden() {
      */
     public DataStream rollover(Index newWriteIndex) {
         assert newWriteIndex.getName().equals(getDefaultBackingIndexName(name, generation + 1));
+        if (replicated) {
+            throw new IllegalArgumentException("data stream [" + name + "] cannot be rolled over, " +
+                "because it is a replicated data stream");
+        }
+
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.add(newWriteIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata, hidden);
+        return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata, hidden, replicated);
     }
 
     /**
@@ -126,7 +144,7 @@ public DataStream removeBackingIndex(Index index) {
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.remove(index);
         assert backingIndices.size() == indices.size() - 1;
-        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden, replicated);
     }
 
     /**
@@ -151,7 +169,11 @@ public DataStream replaceBackingIndex(Index existingBackingIndex, Index newBacki
                 "it is the write index", existingBackingIndex.getName(), name));
         }
         backingIndices.set(backingIndexPosition, newBackingIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden, replicated);
+    }
+
+    public DataStream promoteDataStream() {
+        return new DataStream(name, timeStampField, indices, getGeneration(), metadata, hidden, false);
     }
 
     /**
@@ -169,7 +191,8 @@ public static String getDefaultBackingIndexName(String dataStreamName, long gene
     public DataStream(StreamInput in) throws IOException {
         this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong(),
             in.getVersion().onOrAfter(Version.V_7_11_0) ? in.readMap(): null,
-            in.getVersion().onOrAfter(HIDDEN_VERSION) && in.readBoolean());
+            in.getVersion().onOrAfter(HIDDEN_VERSION) && in.readBoolean(),
+            in.getVersion().onOrAfter(REPLICATED_VERSION) && in.readBoolean());
     }
 
     public static Diff<DataStream> readDiffFrom(StreamInput in) throws IOException {
@@ -188,6 +211,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(HIDDEN_VERSION)) {
             out.writeBoolean(hidden);
         }
+        if (out.getVersion().onOrAfter(REPLICATED_VERSION)) {
+            out.writeBoolean(replicated);
+        }
     }
 
     public static final ParseField NAME_FIELD = new ParseField("name");
@@ -196,11 +222,12 @@ public void writeTo(StreamOutput out) throws IOException {
     public static final ParseField GENERATION_FIELD = new ParseField("generation");
     public static final ParseField METADATA_FIELD = new ParseField("_meta");
     public static final ParseField HIDDEN_FIELD = new ParseField("hidden");
+    public static final ParseField REPLICATED_FIELD = new ParseField("replicated");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
         args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3],
-            (Map<String, Object>) args[4], args[5] != null && (boolean) args[5]));
+            (Map<String, Object>) args[4], args[5] != null && (boolean) args[5], args[6] != null && (boolean) args[6]));
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), NAME_FIELD);
@@ -209,6 +236,7 @@ public void writeTo(StreamOutput out) throws IOException {
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), GENERATION_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
         PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), HIDDEN_FIELD);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), REPLICATED_FIELD);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -226,6 +254,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(METADATA_FIELD.getPreferredName(), metadata);
         }
         builder.field(HIDDEN_FIELD.getPreferredName(), hidden);
+        builder.field(REPLICATED_FIELD.getPreferredName(), replicated);
         builder.endObject();
         return builder;
     }
@@ -239,12 +268,14 @@ public boolean equals(Object o) {
             timeStampField.equals(that.timeStampField) &&
             indices.equals(that.indices) &&
             generation == that.generation &&
-            Objects.equals(metadata, that.metadata);
+            Objects.equals(metadata, that.metadata) &&
+            hidden == that.hidden &&
+            replicated == that.replicated;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, timeStampField, indices, generation, metadata);
+        return Objects.hash(name, timeStampField, indices, generation, metadata, hidden, replicated);
     }
 
     public static final class TimestampField implements Writeable, ToXContentObject {

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java

@@ -185,7 +185,7 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
         dsBackingIndices.add(writeIndex.getIndex());
         boolean hidden = template.getDataStreamTemplate().isHidden();
         DataStream newDataStream = new DataStream(dataStreamName, timestampField, dsBackingIndices, 1L,
-                                                  template.metadata() != null ? Map.copyOf(template.metadata()) : null, hidden);
+            template.metadata() != null ? Map.copyOf(template.metadata()) : null, hidden, false);
         Metadata.Builder builder = Metadata.builder(currentState.metadata()).put(newDataStream);
         logger.info("adding data stream [{}] with write index [{}] and backing indices [{}]", dataStreamName,
             writeIndex.getIndex().getName(),

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -623,7 +623,7 @@ static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metad
             .map(i -> metadata.get(renameIndex(i.getName(), request, true)).getIndex())
             .collect(Collectors.toList());
         return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration(),
-            dataStream.getMetadata(), dataStream.isHidden());
+            dataStream.getMetadata(), dataStream.isHidden(), dataStream.isReplicated());
     }
 
     public static RestoreInProgress updateRestoreStateWithDeletedIndices(RestoreInProgress oldRestore, Set<Index> deletedIndices) {

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java

@@ -536,7 +536,9 @@ public void testRolloverClusterState() throws Exception {
     }
 
     public void testRolloverClusterStateForDataStream() throws Exception {
-        final DataStream dataStream = DataStreamTestHelper.randomInstance();
+        final DataStream dataStream = DataStreamTestHelper.randomInstance()
+            // ensure no replicate data stream
+            .promoteDataStream();
         ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStream.getName() + "*"), null, null, null, null, null,
             new ComposableIndexTemplate.DataStreamTemplate(), null);
         Metadata.Builder builder = Metadata.builder();
@@ -632,7 +634,9 @@ public void testValidation() throws Exception {
         final boolean useDataStream = randomBoolean();
         final Metadata.Builder builder = Metadata.builder();
         if (useDataStream) {
-            DataStream dataStream = DataStreamTestHelper.randomInstance();
+            DataStream dataStream = DataStreamTestHelper.randomInstance()
+                // ensure no replicate data stream
+                .promoteDataStream();
             rolloverTarget = dataStream.getName();
             sourceIndexName = dataStream.getIndices().get(dataStream.getIndices().size() - 1).getName();
             defaultRolloverIndexName = DataStream.getDefaultBackingIndexName(dataStream.getName(), dataStream.getGeneration() + 1);

server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java

@@ -52,7 +52,7 @@ protected DataStream createTestInstance() {
     }
 
     public void testRollover() {
-        DataStream ds = DataStreamTestHelper.randomInstance();
+        DataStream ds = DataStreamTestHelper.randomInstance().promoteDataStream();
         Index newWriteIndex = new Index(getDefaultBackingIndexName(ds.getName(), ds.getGeneration() + 1), UUIDs.randomBase64UUID(random()));
         DataStream rolledDs = ds.rollover(newWriteIndex);
 

server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java

@@ -2127,7 +2127,7 @@ public void testHiddenDataStreams() {
                 .put(index2, false)
                 .put(justAnIndex, false)
                 .put(new DataStream(dataStream1, createTimestampField("@timestamp"),
-                    List.of(index1.getIndex(), index2.getIndex()), 2, Collections.emptyMap(), true))).build();
+                    List.of(index1.getIndex(), index2.getIndex()), 2, Collections.emptyMap(), true, false))).build();
 
         Index[] result = indexNameExpressionResolver.concreteIndices(state, IndicesOptions.strictExpandHidden(), true, "logs-*");
         assertThat(result, arrayContainingInAnyOrder(index1.getIndex(), index2.getIndex(), justAnIndex.getIndex() ));