Skip to content

Converting an Existing Data Stream to a System DataStream is Broken #121392

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Feb 21, 2025
Merged

Converting an Existing Data Stream to a System DataStream is Broken #121392

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
6289eb9
Converting an Existing Data Stream to a System DataStream is Broken
alexey-ivanov-es Jan 31, 2025
5e92c45
Update docs/changelog/121392.yaml
alexey-ivanov-es Jan 31, 2025
a1b6ceb
[CI] Auto commit changes from spotless
Jan 31, 2025
251e88b
Address review comments
alexey-ivanov-es Feb 7, 2025
9a4ee2c
Merge branch 'main' into ES-10526
alexey-ivanov-es Feb 7, 2025
4ef97b4
Change way how indices backing data streams are handled
alexey-ivanov-es Feb 12, 2025
61f9c4f
[CI] Auto commit changes from spotless
Feb 12, 2025
1a7a5d5
break -> return
alexey-ivanov-es Feb 12, 2025
19f9700
Merge branch 'main' into ES-10526
alexey-ivanov-es Feb 13, 2025
78e8059
Migrate SystemIndexMetadataUpgradeService from unbatched tasks to Clu…
alexey-ivanov-es Feb 13, 2025
d20b021
[CI] Auto commit changes from spotless
Feb 13, 2025
ead1cc8
Update wording in changelog
alexey-ivanov-es Feb 13, 2025
4993fdc
Merge branch 'main' into ES-10526
alexey-ivanov-es Feb 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,6 +301,13 @@ public boolean isFailureStoreIndex(String indexName) {
return failureIndices.containsIndex(indexName);
}

/**
* Returns true if the index name provided belongs to this data stream.
*/
public boolean containsIndex(String indexName) {
return backingIndices.containsIndex(indexName) || failureIndices.containsIndex(indexName);
}

public DataStreamOptions getDataStreamOptions() {
return dataStreamOptions;
}
Expand Down Expand Up @@ -1039,7 +1046,7 @@ private boolean isIndexOlderThan(
* we return false.
*/
public boolean isIndexManagedByDataStreamLifecycle(Index index, Function<String, IndexMetadata> indexMetadataSupplier) {
if (backingIndices.containsIndex(index.getName()) == false && failureIndices.containsIndex(index.getName()) == false) {
if (containsIndex(index.getName()) == false) {
return false;
}
IndexMetadata indexMetadata = indexMetadataSupplier.apply(index.getName());
Expand Down
41 changes: 32 additions & 9 deletions ...r/src/main/java/org/elasticsearch/cluster/metadata/SystemIndexMetadataUpgradeService.java
View file
Open in desktop
Copy link
Contributor Author

@alexey-ivanov-es alexey-ivanov-es Jan 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At https://github.com/elastic/elasticsearch/pull/121392/files#diff-1123b904a4743231e7321993f73a75ae603c6658b1e6acf6bf8656b69370a410L70 we currently only check for indices that should become system indices. Should we also check for data streams there?

It seems that if a data stream exists, it always has at least one backing index. If the data stream is non-system, the backing index will also be non-system. Then, when the data stream is marked as system, the index will be picked up by the task, and both will be converted to system ones. Does this approach make sense, or should we explicitly check for data streams as well?

Copy link
Contributor

@gmarouli gmarouli Feb 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, good question. For now that is the case, but we have talked about changing this one day but there are no concrete plans.

If I am not mistaken, this is just looping over one more data set, so I think it would be better to be "thorough" and avoid surprises in the future. What do you think?

Copy link
Contributor Author

@alexey-ivanov-es alexey-ivanov-es Feb 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we have talked about changing this one day

I'll add check for data streams here as well. Thank you!

Original file line number Diff line number Diff line change
Expand Up @@ -129,14 +129,28 @@ private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String
}

// visible for testing
SystemIndexMetadataUpdateTask getTask() {
return new SystemIndexMetadataUpdateTask();
ClusterState executeMetadataUpdateTask(ClusterState clusterState) {
return new SystemIndexMetadataUpdateTask().execute(clusterState);
}

public class SystemIndexMetadataUpdateTask extends ClusterStateUpdateTask {
private class SystemIndexMetadataUpdateTask extends ClusterStateUpdateTask {

@Override
public ClusterState execute(ClusterState currentState) throws Exception {
public ClusterState execute(ClusterState currentState) {
List<IndexMetadata> updatedMetadata = updateIndices(currentState);
List<DataStream> updatedDataStreams = updateDataStreams(currentState);

if (updatedMetadata.isEmpty() == false || updatedDataStreams.isEmpty() == false) {
Metadata.Builder builder = Metadata.builder(currentState.metadata());
updatedMetadata.forEach(idxMeta -> builder.put(idxMeta, true));
updatedDataStreams.forEach(builder::put);

return ClusterState.builder(currentState).metadata(builder).build();
}
return currentState;
}

private List<IndexMetadata> updateIndices(ClusterState currentState) {
final Map<String, IndexMetadata> indexMetadataMap = currentState.metadata().indices();
final List<IndexMetadata> updatedMetadata = new ArrayList<>();
for (Map.Entry<String, IndexMetadata> entry : indexMetadataMap.entrySet()) {
Expand Down Expand Up @@ -172,13 +186,22 @@ public ClusterState execute(ClusterState currentState) throws Exception {
updatedMetadata.add(builder.build());
}
}
return updatedMetadata;
}

if (updatedMetadata.isEmpty() == false) {
final Metadata.Builder builder = Metadata.builder(currentState.metadata());
updatedMetadata.forEach(idxMeta -> builder.put(idxMeta, true));
return ClusterState.builder(currentState).metadata(builder).build();
private List<DataStream> updateDataStreams(ClusterState currentState) {
List<DataStream> updatedDataStreams = new ArrayList<>();
for (DataStream dataStream : currentState.getMetadata().dataStreams().values()) {
if (dataStream.isSystem() == false && systemIndices.isSystemDataStream(dataStream.getName())) {
DataStream updatedDataStream = dataStream.copy()
.setSystem(true)
.setHidden(true)
.build();

updatedDataStreams.add(updatedDataStream);
Copy link
Contributor

@JVerwolf JVerwolf Feb 3, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like the indices version also updates any aliases to be hidden. Do we need to do that here?

Copy link
Contributor Author

@alexey-ivanov-es alexey-ivanov-es Feb 4, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I verified this and data stream aliases become hidden after the data stream is marked as hidden/system. Don't see an easy way to write a test for this, unfortunately

}
}
return currentState;
return updatedDataStreams;
}

@Override
Expand Down
61 changes: 57 additions & 4 deletions .../test/java/org/elasticsearch/cluster/metadata/SystemIndexMetadataUpgradeServiceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,14 @@
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.IndexVersion;
import org.elasticsearch.indices.ExecutorNames;
import org.elasticsearch.indices.SystemDataStreamDescriptor;
import org.elasticsearch.indices.SystemIndexDescriptor;
import org.elasticsearch.indices.SystemIndices;
import org.elasticsearch.test.ESTestCase;
import org.junit.Before;

import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
Expand Down Expand Up @@ -49,13 +52,25 @@ public class SystemIndexMetadataUpgradeServiceTests extends ESTestCase {
.setOrigin("FAKE_ORIGIN")
.build();

private static final String SYSTEM_DATA_STREAM_NAME = ".my-ds";
private static final String SYSTEM_DATA_STREAM_INDEX_NAME = DataStream.BACKING_INDEX_PREFIX + SYSTEM_DATA_STREAM_NAME + "-1";
private static final SystemDataStreamDescriptor SYSTEM_DATA_STREAM_DESCRIPTOR = new SystemDataStreamDescriptor(
SYSTEM_DATA_STREAM_NAME, "System datastream for test", SystemDataStreamDescriptor.Type.INTERNAL,
ComposableIndexTemplate.builder().build(), Collections.emptyMap(), Collections.singletonList("FAKE_ORIGIN"),
ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
);

private SystemIndexMetadataUpgradeService service;

@Before
public void setUpTest() {
// set up a system index upgrade service
this.service = new SystemIndexMetadataUpgradeService(
new SystemIndices(List.of(new SystemIndices.Feature("foo", "a test feature", List.of(DESCRIPTOR)))),
new SystemIndices(List.of(
new SystemIndices.Feature("foo", "a test feature", List.of(DESCRIPTOR)),
new SystemIndices.Feature("sds", "system data stream feature",
Collections.emptyList(), Collections.singletonList(SYSTEM_DATA_STREAM_DESCRIPTOR))
)),
mock(ClusterService.class)
);
}
Expand All @@ -75,6 +90,19 @@ public void testUpgradeVisibleIndexToSystemIndex() throws Exception {
assertSystemUpgradeAppliesHiddenSetting(hiddenIndexMetadata);
}

public void testUpgradeDataStreamToSystemDataStream() {
IndexMetadata dsIndexMetadata = IndexMetadata.builder(SYSTEM_DATA_STREAM_INDEX_NAME)
.system(false)
.settings(getSettingsBuilder().put(IndexMetadata.SETTING_INDEX_HIDDEN, true))
.build();
DataStream dataStream = DataStream.builder(SYSTEM_DATA_STREAM_NAME, Collections.singletonList(dsIndexMetadata.getIndex()))
.setHidden(false)
.setSystem(false)
.build();

assertSystemUpgradeAppliesHiddenSettingForDataStream(dataStream, dsIndexMetadata);
Copy link
Contributor

@JVerwolf JVerwolf Feb 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is only used in one place, should this remain separate?

Copy link
Contributor Author

@alexey-ivanov-es alexey-ivanov-es Feb 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Inlined

}

/**
* If a system index erroneously is set to visible, we should remedy that situation.
*/
Expand Down Expand Up @@ -209,7 +237,7 @@ public void testIsVisible() {
assertThat(service.requiresUpdate(systemVisibleIndex), equalTo(true));
}

private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMetadata) throws Exception {
private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMetadata) {
assertTrue("Metadata should require update but does not", service.requiresUpdate(hiddenIndexMetadata));
Metadata.Builder clusterMetadata = new Metadata.Builder();
clusterMetadata.put(IndexMetadata.builder(hiddenIndexMetadata));
Expand All @@ -220,13 +248,38 @@ private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMe
.build();

// Get a metadata upgrade task and execute it on the initial cluster state
ClusterState newState = service.getTask().execute(clusterState);
ClusterState newState = service.executeMetadataUpdateTask(clusterState);

IndexMetadata result = newState.metadata().index(SYSTEM_INDEX_NAME);
assertThat(result.isSystem(), equalTo(true));
assertThat(result.isHidden(), equalTo(true));
}

private void assertSystemUpgradeAppliesHiddenSettingForDataStream(DataStream shouldBeSystemDataStream, IndexMetadata dsIndexMetadata) {
assertTrue(shouldBeSystemDataStream.containsIndex(dsIndexMetadata.getIndex().getName()));
assertTrue("Metadata should require update but does not", service.requiresUpdate(dsIndexMetadata));

Metadata.Builder clusterMetadata = new Metadata.Builder();
clusterMetadata.put(shouldBeSystemDataStream);
clusterMetadata.put(dsIndexMetadata, true);

ClusterState clusterState = ClusterState.builder(new ClusterName("system-index-metadata-upgrade-service-tests"))
.metadata(clusterMetadata.build())
.customs(Map.of())
.build();

// Execute a metadata upgrade task on the initial cluster state
ClusterState newState = service.executeMetadataUpdateTask(clusterState);

DataStream updatedDataStream = newState.metadata().dataStreams().get(shouldBeSystemDataStream.getName());
assertThat(updatedDataStream.isSystem(), equalTo(true));
assertThat(updatedDataStream.isHidden(), equalTo(true));

IndexMetadata updatedIndexMetadata = newState.metadata().index(dsIndexMetadata.getIndex().getName());
assertThat(updatedIndexMetadata.isSystem(), equalTo(true));
assertThat(updatedIndexMetadata.isHidden(), equalTo(true));
}

private void assertSystemUpgradeHidesAlias(IndexMetadata visibleAliasMetadata) throws Exception {
assertTrue("Metadata should require update but does not", service.requiresUpdate(visibleAliasMetadata));
Metadata.Builder clusterMetadata = new Metadata.Builder();
Expand All @@ -238,7 +291,7 @@ private void assertSystemUpgradeHidesAlias(IndexMetadata visibleAliasMetadata) t
.build();

// Get a metadata upgrade task and execute it on the initial cluster state
ClusterState newState = service.getTask().execute(clusterState);
ClusterState newState = service.executeMetadataUpdateTask(clusterState);

IndexMetadata result = newState.metadata().index(SYSTEM_INDEX_NAME);
assertThat(result.isSystem(), equalTo(true));
Expand Down