|
|
@@ -22,7 +22,6 @@ import org.elasticsearch.cluster.SimpleDiffable;
|
|
|
import org.elasticsearch.cluster.metadata.DataStreamLifecycle.Downsampling.Round;
|
|
|
import org.elasticsearch.common.ParsingException;
|
|
|
import org.elasticsearch.common.Strings;
|
|
|
-import org.elasticsearch.common.TriFunction;
|
|
|
import org.elasticsearch.common.bytes.BytesReference;
|
|
|
import org.elasticsearch.common.io.stream.StreamInput;
|
|
|
import org.elasticsearch.common.io.stream.StreamOutput;
|
|
|
@@ -52,7 +51,6 @@ import java.util.ArrayList;
|
|
|
import java.util.Collection;
|
|
|
import java.util.Comparator;
|
|
|
import java.util.HashMap;
|
|
|
-import java.util.HashSet;
|
|
|
import java.util.List;
|
|
|
import java.util.Locale;
|
|
|
import java.util.Map;
|
|
|
@@ -102,7 +100,6 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
|
|
|
private final LongSupplier timeProvider;
|
|
|
private final String name;
|
|
|
- private final List<Index> indices;
|
|
|
private final long generation;
|
|
|
@Nullable
|
|
|
private final Map<String, Object> metadata;
|
|
|
@@ -114,12 +111,10 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
private final IndexMode indexMode;
|
|
|
@Nullable
|
|
|
private final DataStreamLifecycle lifecycle;
|
|
|
- private final boolean rolloverOnWrite;
|
|
|
private final boolean failureStoreEnabled;
|
|
|
- private final List<Index> failureIndices;
|
|
|
- private volatile Set<String> failureStoreLookup;
|
|
|
- @Nullable
|
|
|
- private final DataStreamAutoShardingEvent autoShardingEvent;
|
|
|
+
|
|
|
+ private final DataStreamIndices backingIndices;
|
|
|
+ private final DataStreamIndices failureIndices;
|
|
|
|
|
|
public DataStream(
|
|
|
String name,
|
|
|
@@ -139,7 +134,6 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
) {
|
|
|
this(
|
|
|
name,
|
|
|
- indices,
|
|
|
generation,
|
|
|
metadata,
|
|
|
hidden,
|
|
|
@@ -150,16 +144,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
indexMode,
|
|
|
lifecycle,
|
|
|
failureStoreEnabled,
|
|
|
- failureIndices,
|
|
|
- rolloverOnWrite,
|
|
|
- autoShardingEvent
|
|
|
+ new DataStreamIndices(BACKING_INDEX_PREFIX, List.copyOf(indices), rolloverOnWrite, autoShardingEvent),
|
|
|
+ new DataStreamIndices(FAILURE_STORE_PREFIX, List.copyOf(failureIndices), false, null)
|
|
|
);
|
|
|
}
|
|
|
|
|
|
// visible for testing
|
|
|
DataStream(
|
|
|
String name,
|
|
|
- List<Index> indices,
|
|
|
long generation,
|
|
|
Map<String, Object> metadata,
|
|
|
boolean hidden,
|
|
|
@@ -170,13 +162,10 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
IndexMode indexMode,
|
|
|
DataStreamLifecycle lifecycle,
|
|
|
boolean failureStoreEnabled,
|
|
|
- List<Index> failureIndices,
|
|
|
- boolean rolloverOnWrite,
|
|
|
- @Nullable DataStreamAutoShardingEvent autoShardingEvent
|
|
|
+ DataStreamIndices backingIndices,
|
|
|
+ DataStreamIndices failureIndices
|
|
|
) {
|
|
|
this.name = name;
|
|
|
- this.indices = List.copyOf(indices);
|
|
|
- assert indices.isEmpty() == false;
|
|
|
this.generation = generation;
|
|
|
this.metadata = metadata;
|
|
|
assert system == false || hidden; // system indices must be hidden
|
|
|
@@ -188,21 +177,11 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
this.indexMode = indexMode;
|
|
|
this.lifecycle = lifecycle;
|
|
|
this.failureStoreEnabled = failureStoreEnabled;
|
|
|
+ assert backingIndices.indices.isEmpty() == false;
|
|
|
+ assert replicated == false || (backingIndices.rolloverOnWrite == false && failureIndices.rolloverOnWrite == false)
|
|
|
+ : "replicated data streams cannot be marked for lazy rollover";
|
|
|
+ this.backingIndices = backingIndices;
|
|
|
this.failureIndices = failureIndices;
|
|
|
- assert assertConsistent(this.indices);
|
|
|
- assert replicated == false || rolloverOnWrite == false : "replicated data streams cannot be marked for lazy rollover";
|
|
|
- this.rolloverOnWrite = rolloverOnWrite;
|
|
|
- this.autoShardingEvent = autoShardingEvent;
|
|
|
- }
|
|
|
-
|
|
|
- private static boolean assertConsistent(List<Index> indices) {
|
|
|
- assert indices.size() > 0;
|
|
|
- final Set<String> indexNames = new HashSet<>();
|
|
|
- for (Index index : indices) {
|
|
|
- final boolean added = indexNames.add(index.getName());
|
|
|
- assert added : "found duplicate index entries in " + indices;
|
|
|
- }
|
|
|
- return true;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
@@ -222,20 +201,16 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
|
|
|
@Override
|
|
|
public List<Index> getIndices() {
|
|
|
- return indices;
|
|
|
+ return backingIndices.indices;
|
|
|
}
|
|
|
|
|
|
public long getGeneration() {
|
|
|
return generation;
|
|
|
}
|
|
|
|
|
|
- public List<Index> getFailureIndices() {
|
|
|
- return failureIndices;
|
|
|
- }
|
|
|
-
|
|
|
@Override
|
|
|
public Index getWriteIndex() {
|
|
|
- return indices.get(indices.size() - 1);
|
|
|
+ return backingIndices.getWriteIndex();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -243,29 +218,18 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
*/
|
|
|
@Nullable
|
|
|
public Index getFailureStoreWriteIndex() {
|
|
|
- return isFailureStoreEnabled() == false || failureIndices.isEmpty() ? null : failureIndices.get(failureIndices.size() - 1);
|
|
|
+ return isFailureStoreEnabled() == false || failureIndices.indices.isEmpty() ? null : failureIndices.getWriteIndex();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Returns true if the index name provided belongs to a failure store index.
|
|
|
- * This method builds a local Set with all the failure store index names and then checks if it contains the name.
|
|
|
- * This will perform better if there are multiple indices of this data stream checked.
|
|
|
*/
|
|
|
public boolean isFailureStoreIndex(String indexName) {
|
|
|
- if (failureStoreLookup == null) {
|
|
|
- // There is a chance this will be calculated twice, but it's a relatively cheap action,
|
|
|
- // so it's not worth synchronising
|
|
|
- if (failureIndices == null || failureIndices.isEmpty()) {
|
|
|
- failureStoreLookup = Set.of();
|
|
|
- } else {
|
|
|
- failureStoreLookup = failureIndices.stream().map(Index::getName).collect(Collectors.toSet());
|
|
|
- }
|
|
|
- }
|
|
|
- return failureStoreLookup.contains(indexName);
|
|
|
+ return failureIndices.containsIndex(indexName);
|
|
|
}
|
|
|
|
|
|
public boolean rolloverOnWrite() {
|
|
|
- return rolloverOnWrite;
|
|
|
+ return backingIndices.rolloverOnWrite;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -275,8 +239,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* an end time that is less than the provided timestamp. Otherwise <code>null</code> is returned.
|
|
|
*/
|
|
|
public Index selectTimeSeriesWriteIndex(Instant timestamp, Metadata metadata) {
|
|
|
- for (int i = indices.size() - 1; i >= 0; i--) {
|
|
|
- Index index = indices.get(i);
|
|
|
+ for (int i = backingIndices.indices.size() - 1; i >= 0; i--) {
|
|
|
+ Index index = backingIndices.indices.get(i);
|
|
|
IndexMetadata im = metadata.index(index);
|
|
|
|
|
|
// TODO: make index_mode, start and end time fields in IndexMetadata class.
|
|
|
@@ -306,7 +270,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
public void validate(Function<String, IndexMetadata> imSupplier) {
|
|
|
if (indexMode == IndexMode.TIME_SERIES) {
|
|
|
// Get a sorted overview of each backing index with there start and end time range:
|
|
|
- var startAndEndTimes = indices.stream().map(index -> {
|
|
|
+ var startAndEndTimes = backingIndices.indices.stream().map(index -> {
|
|
|
IndexMetadata im = imSupplier.apply(index.getName());
|
|
|
if (im == null) {
|
|
|
throw new IllegalStateException("index [" + index.getName() + "] is not found in the index metadata supplier");
|
|
|
@@ -407,7 +371,19 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* Returns the latest auto sharding event that happened for this data stream
|
|
|
*/
|
|
|
public DataStreamAutoShardingEvent getAutoShardingEvent() {
|
|
|
- return autoShardingEvent;
|
|
|
+ return backingIndices.autoShardingEvent;
|
|
|
+ }
|
|
|
+
|
|
|
+ public DataStreamIndices getBackingIndices() {
|
|
|
+ return backingIndices;
|
|
|
+ }
|
|
|
+
|
|
|
+ public DataStreamIndices getFailureIndices() {
|
|
|
+ return failureIndices;
|
|
|
+ }
|
|
|
+
|
|
|
+ public DataStreamIndices getDataStreamIndices(boolean failureStore) {
|
|
|
+ return failureStore ? this.failureIndices : backingIndices;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -446,15 +422,11 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
indexMode = null;
|
|
|
}
|
|
|
|
|
|
- List<Index> backingIndices = new ArrayList<>(indices);
|
|
|
+ List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices);
|
|
|
backingIndices.add(writeIndex);
|
|
|
- return copy().setIndices(backingIndices)
|
|
|
- .setGeneration(generation)
|
|
|
- .setReplicated(false)
|
|
|
- .setIndexMode(indexMode)
|
|
|
- .setAutoShardingEvent(autoShardingEvent)
|
|
|
- .setRolloverOnWrite(false)
|
|
|
- .build();
|
|
|
+ return copy().setBackingIndices(
|
|
|
+ this.backingIndices.copy().setIndices(backingIndices).setAutoShardingEvent(autoShardingEvent).setRolloverOnWrite(false).build()
|
|
|
+ ).setGeneration(generation).setIndexMode(indexMode).build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -475,56 +447,32 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* Like {@link #rolloverFailureStore(Index, long)}, but does no validation, use with care only.
|
|
|
*/
|
|
|
public DataStream unsafeRolloverFailureStore(Index writeIndex, long generation) {
|
|
|
- List<Index> failureIndices = new ArrayList<>(this.failureIndices);
|
|
|
+ List<Index> failureIndices = new ArrayList<>(this.failureIndices.indices);
|
|
|
failureIndices.add(writeIndex);
|
|
|
- return copy().setGeneration(generation).setReplicated(false).setFailureIndices(failureIndices).build();
|
|
|
+ return copy().setGeneration(generation).setFailureIndices(this.failureIndices.copy().setIndices(failureIndices).build()).build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Generates the next write index name and <code>generation</code> to be used for rolling over this data stream.
|
|
|
*
|
|
|
* @param clusterMetadata Cluster metadata
|
|
|
+ * @param dataStreamIndices The data stream indices that we're generating the next write index name and generation for
|
|
|
* @return tuple of the next write index name and next generation.
|
|
|
*/
|
|
|
- public Tuple<String, Long> nextWriteIndexAndGeneration(Metadata clusterMetadata) {
|
|
|
- ensureNotReplicated();
|
|
|
- return unsafeNextWriteIndexAndGeneration(clusterMetadata);
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * Like {@link #nextWriteIndexAndGeneration(Metadata)}, but does no validation, use with care only.
|
|
|
- */
|
|
|
- public Tuple<String, Long> unsafeNextWriteIndexAndGeneration(Metadata clusterMetadata) {
|
|
|
- return generateNextWriteIndexAndGeneration(clusterMetadata, DataStream::getDefaultBackingIndexName);
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * Generates the next write index name and <code>generation</code> to be used for rolling over the failure store of this data stream.
|
|
|
- *
|
|
|
- * @param clusterMetadata Cluster metadata
|
|
|
- * @return tuple of the next failure store write index name and next generation.
|
|
|
- */
|
|
|
- public Tuple<String, Long> nextFailureStoreWriteIndexAndGeneration(Metadata clusterMetadata) {
|
|
|
+ public Tuple<String, Long> nextWriteIndexAndGeneration(Metadata clusterMetadata, DataStreamIndices dataStreamIndices) {
|
|
|
ensureNotReplicated();
|
|
|
- return unsafeNextFailureStoreWriteIndexAndGeneration(clusterMetadata);
|
|
|
+ return unsafeNextWriteIndexAndGeneration(clusterMetadata, dataStreamIndices);
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Like {@link #nextFailureStoreWriteIndexAndGeneration(Metadata)}, but does no validation, use with care only.
|
|
|
+ * Like {@link #nextWriteIndexAndGeneration(Metadata, DataStreamIndices)}, but does no validation, use with care only.
|
|
|
*/
|
|
|
- public Tuple<String, Long> unsafeNextFailureStoreWriteIndexAndGeneration(Metadata clusterMetadata) {
|
|
|
- return generateNextWriteIndexAndGeneration(clusterMetadata, DataStream::getDefaultFailureStoreName);
|
|
|
- }
|
|
|
-
|
|
|
- private Tuple<String, Long> generateNextWriteIndexAndGeneration(
|
|
|
- Metadata clusterMetadata,
|
|
|
- TriFunction<String, Long, Long, String> nameGenerator
|
|
|
- ) {
|
|
|
+ public Tuple<String, Long> unsafeNextWriteIndexAndGeneration(Metadata clusterMetadata, DataStreamIndices dataStreamIndices) {
|
|
|
String newWriteIndexName;
|
|
|
long generation = this.generation;
|
|
|
long currentTimeMillis = timeProvider.getAsLong();
|
|
|
do {
|
|
|
- newWriteIndexName = nameGenerator.apply(getName(), ++generation, currentTimeMillis);
|
|
|
+ newWriteIndexName = dataStreamIndices.generateName(name, ++generation, currentTimeMillis);
|
|
|
} while (clusterMetadata.hasIndexAbstraction(newWriteIndexName));
|
|
|
return Tuple.tuple(newWriteIndexName, generation);
|
|
|
}
|
|
|
@@ -544,14 +492,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* @throws IllegalArgumentException if {@code index} is not a backing index or is the current write index of the data stream
|
|
|
*/
|
|
|
public DataStream removeBackingIndex(Index index) {
|
|
|
- int backingIndexPosition = indices.indexOf(index);
|
|
|
+ int backingIndexPosition = backingIndices.indices.indexOf(index);
|
|
|
|
|
|
if (backingIndexPosition == -1) {
|
|
|
throw new IllegalArgumentException(
|
|
|
String.format(Locale.ROOT, "index [%s] is not part of data stream [%s]", index.getName(), name)
|
|
|
);
|
|
|
}
|
|
|
- if (indices.size() == (backingIndexPosition + 1)) {
|
|
|
+ if (backingIndices.indices.size() == (backingIndexPosition + 1)) {
|
|
|
throw new IllegalArgumentException(
|
|
|
String.format(
|
|
|
Locale.ROOT,
|
|
|
@@ -562,10 +510,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
);
|
|
|
}
|
|
|
|
|
|
- List<Index> backingIndices = new ArrayList<>(indices);
|
|
|
+ List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices);
|
|
|
backingIndices.remove(index);
|
|
|
- assert backingIndices.size() == indices.size() - 1;
|
|
|
- return copy().setIndices(backingIndices).setGeneration(generation + 1).build();
|
|
|
+ assert backingIndices.size() == this.backingIndices.indices.size() - 1;
|
|
|
+ return copy().setBackingIndices(this.backingIndices.copy().setIndices(backingIndices).build())
|
|
|
+ .setGeneration(generation + 1)
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -578,7 +528,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* data stream
|
|
|
*/
|
|
|
public DataStream removeFailureStoreIndex(Index index) {
|
|
|
- int failureIndexPosition = failureIndices.indexOf(index);
|
|
|
+ int failureIndexPosition = failureIndices.indices.indexOf(index);
|
|
|
|
|
|
if (failureIndexPosition == -1) {
|
|
|
throw new IllegalArgumentException(
|
|
|
@@ -588,7 +538,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
|
|
|
// TODO: When failure stores are lazily created, this wont necessarily be required anymore. We can remove the failure store write
|
|
|
// index as long as we mark the data stream to lazily rollover the failure store with no conditions on its next write
|
|
|
- if (failureIndices.size() == (failureIndexPosition + 1)) {
|
|
|
+ if (failureIndices.indices.size() == (failureIndexPosition + 1)) {
|
|
|
throw new IllegalArgumentException(
|
|
|
String.format(
|
|
|
Locale.ROOT,
|
|
|
@@ -599,10 +549,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
);
|
|
|
}
|
|
|
|
|
|
- List<Index> updatedFailureIndices = new ArrayList<>(failureIndices);
|
|
|
+ List<Index> updatedFailureIndices = new ArrayList<>(failureIndices.indices);
|
|
|
updatedFailureIndices.remove(index);
|
|
|
- assert updatedFailureIndices.size() == failureIndices.size() - 1;
|
|
|
- return copy().setGeneration(generation + 1).setFailureIndices(updatedFailureIndices).build();
|
|
|
+ assert updatedFailureIndices.size() == failureIndices.indices.size() - 1;
|
|
|
+ return copy().setFailureIndices(failureIndices.copy().setIndices(updatedFailureIndices).build())
|
|
|
+ .setGeneration(generation + 1)
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -616,14 +568,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* existing index.
|
|
|
*/
|
|
|
public DataStream replaceBackingIndex(Index existingBackingIndex, Index newBackingIndex) {
|
|
|
- List<Index> backingIndices = new ArrayList<>(indices);
|
|
|
+ List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices);
|
|
|
int backingIndexPosition = backingIndices.indexOf(existingBackingIndex);
|
|
|
if (backingIndexPosition == -1) {
|
|
|
throw new IllegalArgumentException(
|
|
|
String.format(Locale.ROOT, "index [%s] is not part of data stream [%s]", existingBackingIndex.getName(), name)
|
|
|
);
|
|
|
}
|
|
|
- if (indices.size() == (backingIndexPosition + 1)) {
|
|
|
+ if (this.backingIndices.indices.size() == (backingIndexPosition + 1)) {
|
|
|
throw new IllegalArgumentException(
|
|
|
String.format(
|
|
|
Locale.ROOT,
|
|
|
@@ -634,7 +586,9 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
);
|
|
|
}
|
|
|
backingIndices.set(backingIndexPosition, newBackingIndex);
|
|
|
- return copy().setIndices(backingIndices).setGeneration(generation + 1).build();
|
|
|
+ return copy().setBackingIndices(this.backingIndices.copy().setIndices(backingIndices).build())
|
|
|
+ .setGeneration(generation + 1)
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -656,10 +610,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
// ensure that no aliases reference index
|
|
|
ensureNoAliasesOnIndex(clusterMetadata, index);
|
|
|
|
|
|
- List<Index> backingIndices = new ArrayList<>(indices);
|
|
|
+ List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices);
|
|
|
backingIndices.add(0, index);
|
|
|
- assert backingIndices.size() == indices.size() + 1;
|
|
|
- return copy().setIndices(backingIndices).setGeneration(generation + 1).build();
|
|
|
+ assert backingIndices.size() == this.backingIndices.indices.size() + 1;
|
|
|
+ return copy().setBackingIndices(this.backingIndices.copy().setIndices(backingIndices).build())
|
|
|
+ .setGeneration(generation + 1)
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -680,10 +636,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
|
|
|
ensureNoAliasesOnIndex(clusterMetadata, index);
|
|
|
|
|
|
- List<Index> updatedFailureIndices = new ArrayList<>(failureIndices);
|
|
|
+ List<Index> updatedFailureIndices = new ArrayList<>(failureIndices.indices);
|
|
|
updatedFailureIndices.add(0, index);
|
|
|
- assert updatedFailureIndices.size() == failureIndices.size() + 1;
|
|
|
- return copy().setGeneration(generation + 1).setFailureIndices(updatedFailureIndices).build();
|
|
|
+ assert updatedFailureIndices.size() == failureIndices.indices.size() + 1;
|
|
|
+ return copy().setFailureIndices(failureIndices.copy().setIndices(updatedFailureIndices).build())
|
|
|
+ .setGeneration(generation + 1)
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -742,7 +700,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
@Nullable
|
|
|
public DataStream snapshot(Collection<String> indicesInSnapshot) {
|
|
|
// do not include indices not available in the snapshot
|
|
|
- List<Index> reconciledIndices = new ArrayList<>(this.indices);
|
|
|
+ List<Index> reconciledIndices = new ArrayList<>(this.backingIndices.indices);
|
|
|
if (reconciledIndices.removeIf(x -> indicesInSnapshot.contains(x.getName()) == false) == false) {
|
|
|
return this;
|
|
|
}
|
|
|
@@ -751,7 +709,9 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
return null;
|
|
|
}
|
|
|
|
|
|
- return copy().setIndices(reconciledIndices).setMetadata(metadata == null ? null : new HashMap<>(metadata)).build();
|
|
|
+ return copy().setBackingIndices(backingIndices.copy().setIndices(reconciledIndices).build())
|
|
|
+ .setMetadata(metadata == null ? null : new HashMap<>(metadata))
|
|
|
+ .build();
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -792,7 +752,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
Function<String, IndexMetadata> indexMetadataSupplier,
|
|
|
LongSupplier nowSupplier
|
|
|
) {
|
|
|
- assert indices.contains(index) : "the provided index must be a backing index for this datastream";
|
|
|
+ assert backingIndices.indices.contains(index) : "the provided index must be a backing index for this datastream";
|
|
|
if (lifecycle == null || lifecycle.getDownsamplingRounds() == null) {
|
|
|
return List.of();
|
|
|
}
|
|
|
@@ -831,7 +791,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
LongSupplier nowSupplier
|
|
|
) {
|
|
|
List<Index> olderIndices = new ArrayList<>();
|
|
|
- for (Index index : indices) {
|
|
|
+ for (Index index : backingIndices.indices) {
|
|
|
if (isIndexOderThan(index, retentionPeriod.getMillis(), nowSupplier.getAsLong(), indicesPredicate, indexMetadataSupplier)) {
|
|
|
olderIndices.add(index);
|
|
|
}
|
|
|
@@ -864,7 +824,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* we return false.
|
|
|
*/
|
|
|
public boolean isIndexManagedByDataStreamLifecycle(Index index, Function<String, IndexMetadata> indexMetadataSupplier) {
|
|
|
- if (indices.contains(index) == false) {
|
|
|
+ if (backingIndices.indices.contains(index) == false) {
|
|
|
return false;
|
|
|
}
|
|
|
IndexMetadata indexMetadata = indexMetadataSupplier.apply(index.getName());
|
|
|
@@ -936,13 +896,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* @return backing index name
|
|
|
*/
|
|
|
public static String getDefaultBackingIndexName(String dataStreamName, long generation, long epochMillis) {
|
|
|
- return String.format(
|
|
|
- Locale.ROOT,
|
|
|
- BACKING_INDEX_PREFIX + "%s-%s-%06d",
|
|
|
- dataStreamName,
|
|
|
- DATE_FORMATTER.formatMillis(epochMillis),
|
|
|
- generation
|
|
|
- );
|
|
|
+ return getDefaultIndexName(BACKING_INDEX_PREFIX, dataStreamName, generation, epochMillis);
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -955,33 +909,65 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
* @return backing index name
|
|
|
*/
|
|
|
public static String getDefaultFailureStoreName(String dataStreamName, long generation, long epochMillis) {
|
|
|
- return String.format(
|
|
|
- Locale.ROOT,
|
|
|
- FAILURE_STORE_PREFIX + "%s-%s-%06d",
|
|
|
- dataStreamName,
|
|
|
- DATE_FORMATTER.formatMillis(epochMillis),
|
|
|
- generation
|
|
|
- );
|
|
|
+ return getDefaultIndexName(FAILURE_STORE_PREFIX, dataStreamName, generation, epochMillis);
|
|
|
}
|
|
|
|
|
|
- public DataStream(StreamInput in) throws IOException {
|
|
|
- this(
|
|
|
- readName(in),
|
|
|
- readIndices(in),
|
|
|
- in.readVLong(),
|
|
|
- in.readGenericMap(),
|
|
|
- in.readBoolean(),
|
|
|
- in.readBoolean(),
|
|
|
- in.readBoolean(),
|
|
|
- in.getTransportVersion().onOrAfter(TransportVersions.V_8_0_0) ? in.readBoolean() : false,
|
|
|
- in.getTransportVersion().onOrAfter(TransportVersions.V_8_1_0) ? in.readOptionalEnum(IndexMode.class) : null,
|
|
|
- in.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X) ? in.readOptionalWriteable(DataStreamLifecycle::new) : null,
|
|
|
- in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION) ? in.readBoolean() : false,
|
|
|
- in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION) ? readIndices(in) : List.of(),
|
|
|
- in.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0) ? in.readBoolean() : false,
|
|
|
- in.getTransportVersion().onOrAfter(DataStream.ADDED_AUTO_SHARDING_EVENT_VERSION)
|
|
|
- ? in.readOptionalWriteable(DataStreamAutoShardingEvent::new)
|
|
|
- : null
|
|
|
+ /**
|
|
|
+ * Generates the name of the index that conforms to the default naming convention for indices
|
|
|
+ * on data streams given the specified prefix, data stream name, generation, and time.
|
|
|
+ *
|
|
|
+ * @param prefix the prefix that the index name should have
|
|
|
+ * @param dataStreamName name of the data stream
|
|
|
+ * @param generation generation of the data stream
|
|
|
+ * @param epochMillis creation time for the backing index
|
|
|
+ * @return backing index name
|
|
|
+ */
|
|
|
+ private static String getDefaultIndexName(String prefix, String dataStreamName, long generation, long epochMillis) {
|
|
|
+ return String.format(Locale.ROOT, prefix + "%s-%s-%06d", dataStreamName, DATE_FORMATTER.formatMillis(epochMillis), generation);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static DataStream read(StreamInput in) throws IOException {
|
|
|
+ var name = readName(in);
|
|
|
+ var backingIndicesBuilder = DataStreamIndices.backingIndicesBuilder(readIndices(in));
|
|
|
+ var generation = in.readVLong();
|
|
|
+ var metadata = in.readGenericMap();
|
|
|
+ var hidden = in.readBoolean();
|
|
|
+ var replicated = in.readBoolean();
|
|
|
+ var system = in.readBoolean();
|
|
|
+ var allowCustomRouting = in.getTransportVersion().onOrAfter(TransportVersions.V_8_0_0) ? in.readBoolean() : false;
|
|
|
+ var indexMode = in.getTransportVersion().onOrAfter(TransportVersions.V_8_1_0) ? in.readOptionalEnum(IndexMode.class) : null;
|
|
|
+ var lifecycle = in.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)
|
|
|
+ ? in.readOptionalWriteable(DataStreamLifecycle::new)
|
|
|
+ : null;
|
|
|
+ var failureStoreEnabled = in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)
|
|
|
+ ? in.readBoolean()
|
|
|
+ : false;
|
|
|
+ var failureIndices = in.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)
|
|
|
+ ? readIndices(in)
|
|
|
+ : List.<Index>of();
|
|
|
+ var failureIndicesBuilder = DataStreamIndices.failureIndicesBuilder(failureIndices);
|
|
|
+ backingIndicesBuilder.setRolloverOnWrite(in.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0) ? in.readBoolean() : false);
|
|
|
+ if (in.getTransportVersion().onOrAfter(DataStream.ADDED_AUTO_SHARDING_EVENT_VERSION)) {
|
|
|
+ backingIndicesBuilder.setAutoShardingEvent(in.readOptionalWriteable(DataStreamAutoShardingEvent::new));
|
|
|
+ }
|
|
|
+ if (in.getTransportVersion().onOrAfter(TransportVersions.FAILURE_STORE_FIELD_PARITY)) {
|
|
|
+ failureIndicesBuilder.setRolloverOnWrite(in.readBoolean())
|
|
|
+ .setAutoShardingEvent(in.readOptionalWriteable(DataStreamAutoShardingEvent::new));
|
|
|
+ }
|
|
|
+ return new DataStream(
|
|
|
+ name,
|
|
|
+ generation,
|
|
|
+ metadata,
|
|
|
+ hidden,
|
|
|
+ replicated,
|
|
|
+ system,
|
|
|
+ System::currentTimeMillis,
|
|
|
+ allowCustomRouting,
|
|
|
+ indexMode,
|
|
|
+ lifecycle,
|
|
|
+ failureStoreEnabled,
|
|
|
+ backingIndicesBuilder.build(),
|
|
|
+ failureIndicesBuilder.build()
|
|
|
);
|
|
|
}
|
|
|
|
|
|
@@ -996,14 +982,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
}
|
|
|
|
|
|
public static Diff<DataStream> readDiffFrom(StreamInput in) throws IOException {
|
|
|
- return SimpleDiffable.readDiffFrom(DataStream::new, in);
|
|
|
+ return SimpleDiffable.readDiffFrom(DataStream::read, in);
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void writeTo(StreamOutput out) throws IOException {
|
|
|
out.writeString(name);
|
|
|
out.writeString(TIMESTAMP_FIELD_NAME); // TODO: clear this out in the future https://github.com/elastic/elasticsearch/issues/101991
|
|
|
- out.writeCollection(indices);
|
|
|
+ out.writeCollection(backingIndices.indices);
|
|
|
out.writeVLong(generation);
|
|
|
out.writeGenericMap(metadata);
|
|
|
out.writeBoolean(hidden);
|
|
|
@@ -1020,13 +1006,17 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
}
|
|
|
if (out.getTransportVersion().onOrAfter(DataStream.ADDED_FAILURE_STORE_TRANSPORT_VERSION)) {
|
|
|
out.writeBoolean(failureStoreEnabled);
|
|
|
- out.writeCollection(failureIndices);
|
|
|
+ out.writeCollection(failureIndices.indices);
|
|
|
}
|
|
|
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0)) {
|
|
|
- out.writeBoolean(rolloverOnWrite);
|
|
|
+ out.writeBoolean(backingIndices.rolloverOnWrite);
|
|
|
}
|
|
|
if (out.getTransportVersion().onOrAfter(DataStream.ADDED_AUTO_SHARDING_EVENT_VERSION)) {
|
|
|
- out.writeOptionalWriteable(autoShardingEvent);
|
|
|
+ out.writeOptionalWriteable(backingIndices.autoShardingEvent);
|
|
|
+ }
|
|
|
+ if (out.getTransportVersion().onOrAfter(TransportVersions.FAILURE_STORE_FIELD_PARITY)) {
|
|
|
+ out.writeBoolean(failureIndices.rolloverOnWrite);
|
|
|
+ out.writeOptionalWriteable(failureIndices.autoShardingEvent);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -1045,30 +1035,41 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
public static final ParseField FAILURE_INDICES_FIELD = new ParseField("failure_indices");
|
|
|
public static final ParseField ROLLOVER_ON_WRITE_FIELD = new ParseField("rollover_on_write");
|
|
|
public static final ParseField AUTO_SHARDING_FIELD = new ParseField("auto_sharding");
|
|
|
+ public static final ParseField FAILURE_ROLLOVER_ON_WRITE_FIELD = new ParseField("failure_rollover_on_write");
|
|
|
+ public static final ParseField FAILURE_AUTO_SHARDING_FIELD = new ParseField("failure_auto_sharding");
|
|
|
|
|
|
@SuppressWarnings("unchecked")
|
|
|
private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream", args -> {
|
|
|
// Fields behind a feature flag need to be parsed last otherwise the parser will fail when the feature flag is disabled.
|
|
|
// Until the feature flag is removed we keep them separately to be mindful of this.
|
|
|
boolean failureStoreEnabled = DataStream.isFailureStoreFeatureFlagEnabled() && args[12] != null && (boolean) args[12];
|
|
|
- List<Index> failureStoreIndices = DataStream.isFailureStoreFeatureFlagEnabled() && args[13] != null
|
|
|
- ? (List<Index>) args[13]
|
|
|
- : List.of();
|
|
|
+ DataStreamIndices failureIndices = DataStream.isFailureStoreFeatureFlagEnabled()
|
|
|
+ ? new DataStreamIndices(
|
|
|
+ FAILURE_STORE_PREFIX,
|
|
|
+ args[13] != null ? (List<Index>) args[13] : List.of(),
|
|
|
+ args[14] != null && (boolean) args[14],
|
|
|
+ (DataStreamAutoShardingEvent) args[15]
|
|
|
+ )
|
|
|
+ : new DataStreamIndices(FAILURE_STORE_PREFIX, List.of(), false, null);
|
|
|
return new DataStream(
|
|
|
(String) args[0],
|
|
|
- (List<Index>) args[1],
|
|
|
(Long) args[2],
|
|
|
(Map<String, Object>) args[3],
|
|
|
args[4] != null && (boolean) args[4],
|
|
|
args[5] != null && (boolean) args[5],
|
|
|
args[6] != null && (boolean) args[6],
|
|
|
+ System::currentTimeMillis,
|
|
|
args[7] != null && (boolean) args[7],
|
|
|
args[8] != null ? IndexMode.fromString((String) args[8]) : null,
|
|
|
(DataStreamLifecycle) args[9],
|
|
|
failureStoreEnabled,
|
|
|
- failureStoreIndices,
|
|
|
- args[10] != null && (boolean) args[10],
|
|
|
- (DataStreamAutoShardingEvent) args[11]
|
|
|
+ new DataStreamIndices(
|
|
|
+ BACKING_INDEX_PREFIX,
|
|
|
+ (List<Index>) args[1],
|
|
|
+ args[10] != null && (boolean) args[10],
|
|
|
+ (DataStreamAutoShardingEvent) args[11]
|
|
|
+ ),
|
|
|
+ failureIndices
|
|
|
);
|
|
|
});
|
|
|
|
|
|
@@ -1105,6 +1106,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
(p, c) -> Index.fromXContent(p),
|
|
|
FAILURE_INDICES_FIELD
|
|
|
);
|
|
|
+ PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), FAILURE_ROLLOVER_ON_WRITE_FIELD);
|
|
|
+ PARSER.declareObject(
|
|
|
+ ConstructingObjectParser.optionalConstructorArg(),
|
|
|
+ (p, c) -> DataStreamAutoShardingEvent.fromXContent(p),
|
|
|
+ FAILURE_AUTO_SHARDING_FIELD
|
|
|
+ );
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -1132,11 +1139,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
.startObject()
|
|
|
.field(NAME_FIELD.getPreferredName(), TIMESTAMP_FIELD_NAME)
|
|
|
.endObject();
|
|
|
- builder.xContentList(INDICES_FIELD.getPreferredName(), indices);
|
|
|
+ builder.xContentList(INDICES_FIELD.getPreferredName(), backingIndices.indices);
|
|
|
builder.field(GENERATION_FIELD.getPreferredName(), generation);
|
|
|
- if (DataStream.isFailureStoreFeatureFlagEnabled() && failureIndices.isEmpty() == false) {
|
|
|
- builder.xContentList(FAILURE_INDICES_FIELD.getPreferredName(), failureIndices);
|
|
|
- }
|
|
|
if (metadata != null) {
|
|
|
builder.field(METADATA_FIELD.getPreferredName(), metadata);
|
|
|
}
|
|
|
@@ -1146,6 +1150,15 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
builder.field(ALLOW_CUSTOM_ROUTING.getPreferredName(), allowCustomRouting);
|
|
|
if (DataStream.isFailureStoreFeatureFlagEnabled()) {
|
|
|
builder.field(FAILURE_STORE_FIELD.getPreferredName(), failureStoreEnabled);
|
|
|
+ if (failureIndices.indices.isEmpty() == false) {
|
|
|
+ builder.xContentList(FAILURE_INDICES_FIELD.getPreferredName(), failureIndices.indices);
|
|
|
+ }
|
|
|
+ builder.field(FAILURE_ROLLOVER_ON_WRITE_FIELD.getPreferredName(), failureIndices.rolloverOnWrite);
|
|
|
+ if (failureIndices.autoShardingEvent != null) {
|
|
|
+ builder.startObject(FAILURE_AUTO_SHARDING_FIELD.getPreferredName());
|
|
|
+ failureIndices.autoShardingEvent.toXContent(builder, params);
|
|
|
+ builder.endObject();
|
|
|
+ }
|
|
|
}
|
|
|
if (indexMode != null) {
|
|
|
builder.field(INDEX_MODE.getPreferredName(), indexMode);
|
|
|
@@ -1154,10 +1167,10 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
builder.field(LIFECYCLE.getPreferredName());
|
|
|
lifecycle.toXContent(builder, params, rolloverConfiguration, isSystem() ? null : globalRetention);
|
|
|
}
|
|
|
- builder.field(ROLLOVER_ON_WRITE_FIELD.getPreferredName(), rolloverOnWrite);
|
|
|
- if (autoShardingEvent != null) {
|
|
|
+ builder.field(ROLLOVER_ON_WRITE_FIELD.getPreferredName(), backingIndices.rolloverOnWrite);
|
|
|
+ if (backingIndices.autoShardingEvent != null) {
|
|
|
builder.startObject(AUTO_SHARDING_FIELD.getPreferredName());
|
|
|
- autoShardingEvent.toXContent(builder, params);
|
|
|
+ backingIndices.autoShardingEvent.toXContent(builder, params);
|
|
|
builder.endObject();
|
|
|
}
|
|
|
builder.endObject();
|
|
|
@@ -1170,7 +1183,6 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
if (o == null || getClass() != o.getClass()) return false;
|
|
|
DataStream that = (DataStream) o;
|
|
|
return name.equals(that.name)
|
|
|
- && indices.equals(that.indices)
|
|
|
&& generation == that.generation
|
|
|
&& Objects.equals(metadata, that.metadata)
|
|
|
&& hidden == that.hidden
|
|
|
@@ -1180,16 +1192,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
&& indexMode == that.indexMode
|
|
|
&& Objects.equals(lifecycle, that.lifecycle)
|
|
|
&& failureStoreEnabled == that.failureStoreEnabled
|
|
|
- && failureIndices.equals(that.failureIndices)
|
|
|
- && rolloverOnWrite == that.rolloverOnWrite
|
|
|
- && Objects.equals(autoShardingEvent, that.autoShardingEvent);
|
|
|
+ && Objects.equals(backingIndices, that.backingIndices)
|
|
|
+ && Objects.equals(failureIndices, that.failureIndices);
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public int hashCode() {
|
|
|
return Objects.hash(
|
|
|
name,
|
|
|
- indices,
|
|
|
generation,
|
|
|
metadata,
|
|
|
hidden,
|
|
|
@@ -1199,9 +1209,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
indexMode,
|
|
|
lifecycle,
|
|
|
failureStoreEnabled,
|
|
|
- failureIndices,
|
|
|
- rolloverOnWrite,
|
|
|
- autoShardingEvent
|
|
|
+ backingIndices,
|
|
|
+ failureIndices
|
|
|
);
|
|
|
}
|
|
|
|
|
|
@@ -1345,14 +1354,143 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
return new Builder(name, indices);
|
|
|
}
|
|
|
|
|
|
+ public static Builder builder(String name, DataStreamIndices backingIndices) {
|
|
|
+ return new Builder(name, backingIndices);
|
|
|
+ }
|
|
|
+
|
|
|
public Builder copy() {
|
|
|
return new Builder(this);
|
|
|
}
|
|
|
|
|
|
+ public static class DataStreamIndices {
|
|
|
+ private final String namePrefix;
|
|
|
+ private final List<Index> indices;
|
|
|
+ private final boolean rolloverOnWrite;
|
|
|
+ @Nullable
|
|
|
+ private final DataStreamAutoShardingEvent autoShardingEvent;
|
|
|
+ private Set<String> lookup;
|
|
|
+
|
|
|
+ protected DataStreamIndices(
|
|
|
+ String namePrefix,
|
|
|
+ List<Index> indices,
|
|
|
+ boolean rolloverOnWrite,
|
|
|
+ DataStreamAutoShardingEvent autoShardingEvent
|
|
|
+ ) {
|
|
|
+ this.namePrefix = namePrefix;
|
|
|
+ // The list of indices is expected to be an immutable list. We don't create an immutable copy here, as it might have
|
|
|
+ // impact on the performance on some usages.
|
|
|
+ this.indices = indices;
|
|
|
+ this.rolloverOnWrite = rolloverOnWrite;
|
|
|
+ this.autoShardingEvent = autoShardingEvent;
|
|
|
+
|
|
|
+ assert getLookup().size() == indices.size() : "found duplicate index entries in " + indices;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Set<String> getLookup() {
|
|
|
+ if (lookup == null) {
|
|
|
+ lookup = indices.stream().map(Index::getName).collect(Collectors.toSet());
|
|
|
+ }
|
|
|
+ return lookup;
|
|
|
+ }
|
|
|
+
|
|
|
+ public Index getWriteIndex() {
|
|
|
+ return indices.get(indices.size() - 1);
|
|
|
+ }
|
|
|
+
|
|
|
+ public boolean containsIndex(String index) {
|
|
|
+ return getLookup().contains(index);
|
|
|
+ }
|
|
|
+
|
|
|
+ private String generateName(String dataStreamName, long generation, long epochMillis) {
|
|
|
+ return getDefaultIndexName(namePrefix, dataStreamName, generation, epochMillis);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static Builder backingIndicesBuilder(List<Index> indices) {
|
|
|
+ return new Builder(BACKING_INDEX_PREFIX, indices);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static Builder failureIndicesBuilder(List<Index> indices) {
|
|
|
+ return new Builder(FAILURE_STORE_PREFIX, indices);
|
|
|
+ }
|
|
|
+
|
|
|
+ public Builder copy() {
|
|
|
+ return new Builder(this);
|
|
|
+ }
|
|
|
+
|
|
|
+ public List<Index> getIndices() {
|
|
|
+ return indices;
|
|
|
+ }
|
|
|
+
|
|
|
+ public boolean isRolloverOnWrite() {
|
|
|
+ return rolloverOnWrite;
|
|
|
+ }
|
|
|
+
|
|
|
+ public DataStreamAutoShardingEvent getAutoShardingEvent() {
|
|
|
+ return autoShardingEvent;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public boolean equals(Object o) {
|
|
|
+ if (this == o) return true;
|
|
|
+ if (o == null || getClass() != o.getClass()) return false;
|
|
|
+ DataStreamIndices that = (DataStreamIndices) o;
|
|
|
+ return rolloverOnWrite == that.rolloverOnWrite
|
|
|
+ && Objects.equals(namePrefix, that.namePrefix)
|
|
|
+ && Objects.equals(indices, that.indices)
|
|
|
+ && Objects.equals(autoShardingEvent, that.autoShardingEvent);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public int hashCode() {
|
|
|
+ return Objects.hash(namePrefix, indices, rolloverOnWrite, autoShardingEvent);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static class Builder {
|
|
|
+ private final String namePrefix;
|
|
|
+ private List<Index> indices;
|
|
|
+ private boolean rolloverOnWrite = false;
|
|
|
+ @Nullable
|
|
|
+ private DataStreamAutoShardingEvent autoShardingEvent = null;
|
|
|
+
|
|
|
+ private Builder(String namePrefix, List<Index> indices) {
|
|
|
+ this.namePrefix = namePrefix;
|
|
|
+ this.indices = indices;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Builder(DataStreamIndices dataStreamIndices) {
|
|
|
+ this.namePrefix = dataStreamIndices.namePrefix;
|
|
|
+ this.indices = dataStreamIndices.indices;
|
|
|
+ this.rolloverOnWrite = dataStreamIndices.rolloverOnWrite;
|
|
|
+ this.autoShardingEvent = dataStreamIndices.autoShardingEvent;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Set the list of indices. We always create an immutable copy as that's what the constructor expects.
|
|
|
+ */
|
|
|
+ public Builder setIndices(List<Index> indices) {
|
|
|
+ this.indices = List.copyOf(indices);
|
|
|
+ return this;
|
|
|
+ }
|
|
|
+
|
|
|
+ public Builder setRolloverOnWrite(boolean rolloverOnWrite) {
|
|
|
+ this.rolloverOnWrite = rolloverOnWrite;
|
|
|
+ return this;
|
|
|
+ }
|
|
|
+
|
|
|
+ public Builder setAutoShardingEvent(DataStreamAutoShardingEvent autoShardingEvent) {
|
|
|
+ this.autoShardingEvent = autoShardingEvent;
|
|
|
+ return this;
|
|
|
+ }
|
|
|
+
|
|
|
+ public DataStreamIndices build() {
|
|
|
+ return new DataStreamIndices(namePrefix, indices, rolloverOnWrite, autoShardingEvent);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
public static class Builder {
|
|
|
private LongSupplier timeProvider = System::currentTimeMillis;
|
|
|
private String name;
|
|
|
- private List<Index> indices;
|
|
|
private long generation = 1;
|
|
|
@Nullable
|
|
|
private Map<String, Object> metadata = null;
|
|
|
@@ -1364,22 +1502,23 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
private IndexMode indexMode = null;
|
|
|
@Nullable
|
|
|
private DataStreamLifecycle lifecycle = null;
|
|
|
- private boolean rolloverOnWrite = false;
|
|
|
private boolean failureStoreEnabled = false;
|
|
|
- private List<Index> failureIndices = List.of();
|
|
|
- @Nullable
|
|
|
- private DataStreamAutoShardingEvent autoShardingEvent = null;
|
|
|
+ private DataStreamIndices backingIndices;
|
|
|
+ private DataStreamIndices failureIndices = DataStreamIndices.failureIndicesBuilder(List.of()).build();
|
|
|
|
|
|
- public Builder(String name, List<Index> indices) {
|
|
|
+ private Builder(String name, List<Index> indices) {
|
|
|
+ this(name, DataStreamIndices.backingIndicesBuilder(indices).build());
|
|
|
+ }
|
|
|
+
|
|
|
+ private Builder(String name, DataStreamIndices backingIndices) {
|
|
|
this.name = name;
|
|
|
- assert indices.isEmpty() == false : "Cannot create data stream with empty backing indices";
|
|
|
- this.indices = indices;
|
|
|
+ assert backingIndices.indices.isEmpty() == false : "Cannot create data stream with empty backing indices";
|
|
|
+ this.backingIndices = backingIndices;
|
|
|
}
|
|
|
|
|
|
- public Builder(DataStream dataStream) {
|
|
|
+ private Builder(DataStream dataStream) {
|
|
|
timeProvider = dataStream.timeProvider;
|
|
|
name = dataStream.name;
|
|
|
- indices = dataStream.indices;
|
|
|
generation = dataStream.generation;
|
|
|
metadata = dataStream.metadata;
|
|
|
hidden = dataStream.hidden;
|
|
|
@@ -1388,10 +1527,9 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
allowCustomRouting = dataStream.allowCustomRouting;
|
|
|
indexMode = dataStream.indexMode;
|
|
|
lifecycle = dataStream.lifecycle;
|
|
|
- rolloverOnWrite = dataStream.rolloverOnWrite;
|
|
|
failureStoreEnabled = dataStream.failureStoreEnabled;
|
|
|
+ backingIndices = dataStream.backingIndices;
|
|
|
failureIndices = dataStream.failureIndices;
|
|
|
- autoShardingEvent = dataStream.autoShardingEvent;
|
|
|
}
|
|
|
|
|
|
public Builder setTimeProvider(LongSupplier timeProvider) {
|
|
|
@@ -1404,12 +1542,6 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
- public Builder setIndices(List<Index> indices) {
|
|
|
- assert indices.isEmpty() == false : "Cannot create data stream with empty backing indices";
|
|
|
- this.indices = indices;
|
|
|
- return this;
|
|
|
- }
|
|
|
-
|
|
|
public Builder setGeneration(long generation) {
|
|
|
this.generation = generation;
|
|
|
return this;
|
|
|
@@ -1450,30 +1582,34 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
- public Builder setRolloverOnWrite(boolean rolloverOnWrite) {
|
|
|
- this.rolloverOnWrite = rolloverOnWrite;
|
|
|
+ public Builder setFailureStoreEnabled(boolean failureStoreEnabled) {
|
|
|
+ this.failureStoreEnabled = failureStoreEnabled;
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
- public Builder setFailureStoreEnabled(boolean failureStoreEnabled) {
|
|
|
- this.failureStoreEnabled = failureStoreEnabled;
|
|
|
+ public Builder setBackingIndices(DataStreamIndices backingIndices) {
|
|
|
+ assert backingIndices.indices.isEmpty() == false : "Cannot create data stream with empty backing indices";
|
|
|
+ this.backingIndices = backingIndices;
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
- public Builder setFailureIndices(List<Index> failureIndices) {
|
|
|
+ public Builder setFailureIndices(DataStreamIndices failureIndices) {
|
|
|
this.failureIndices = failureIndices;
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
- public Builder setAutoShardingEvent(DataStreamAutoShardingEvent autoShardingEvent) {
|
|
|
- this.autoShardingEvent = autoShardingEvent;
|
|
|
+ public Builder setDataStreamIndices(boolean targetFailureStore, DataStreamIndices indices) {
|
|
|
+ if (targetFailureStore) {
|
|
|
+ setFailureIndices(indices);
|
|
|
+ } else {
|
|
|
+ setBackingIndices(indices);
|
|
|
+ }
|
|
|
return this;
|
|
|
}
|
|
|
|
|
|
public DataStream build() {
|
|
|
return new DataStream(
|
|
|
name,
|
|
|
- indices,
|
|
|
generation,
|
|
|
metadata,
|
|
|
hidden,
|
|
|
@@ -1484,9 +1620,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
|
|
|
indexMode,
|
|
|
lifecycle,
|
|
|
failureStoreEnabled,
|
|
|
- failureIndices,
|
|
|
- rolloverOnWrite,
|
|
|
- autoShardingEvent
|
|
|
+ backingIndices,
|
|
|
+ failureIndices
|
|
|
);
|
|
|
}
|
|
|
}
|
Niels Bauman