Ignore closed indices for reindex (#120244)

Author: masseyke

docs/changelog/120244.yaml

@@ -0,0 +1,5 @@
+pr: 120244
+summary: Ignore closed indices for reindex
+area: Data streams
+type: enhancement
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/deprecation/DeprecatedIndexPredicate.java

@@ -36,6 +36,7 @@ public static Predicate<Index> getReindexRequiredPredicate(Metadata metadata) {
  public static boolean reindexRequired(IndexMetadata indexMetadata) {
  return creationVersionBeforeMinimumWritableVersion(indexMetadata)
  && isNotSearchableSnapshot(indexMetadata)
+ && isNotClosed(indexMetadata)
  && isNotVerifiedReadOnly(indexMetadata);
  }
 
@@ -52,4 +53,8 @@ private static boolean creationVersionBeforeMinimumWritableVersion(IndexMetadata
  return metadata.getCreationVersion().before(MINIMUM_WRITEABLE_VERSION_AFTER_UPGRADE);
  }
 
+ private static boolean isNotClosed(IndexMetadata indexMetadata) {
+ return indexMetadata.getState().equals(IndexMetadata.State.CLOSE) == false;
+ }
+
 }

x-pack/plugin/deprecation/src/test/java/org/elasticsearch/xpack/deprecation/DataStreamDeprecationChecksTests.java

@@ -41,56 +41,79 @@ public void testOldIndicesCheck() {
  int oldIndexCount = randomIntBetween(1, 100);
  int newIndexCount = randomIntBetween(1, 100);
 
- List<Index> allIndices = new ArrayList<>();
  Map<String, IndexMetadata> nameToIndexMetadata = new HashMap<>();
  Set<String> expectedIndices = new HashSet<>();
 
- for (int i = 0; i < oldIndexCount; i++) {
- Settings.Builder settings = settings(IndexVersion.fromId(7170099));
+ DataStream dataStream = createTestDataStream(oldIndexCount, 0, newIndexCount, 0, nameToIndexMetadata, expectedIndices);
 
- String indexName = "old-data-stream-index-" + i;
- if (expectedIndices.isEmpty() == false && randomIntBetween(0, 2) == 0) {
- settings.put(INDEX_STORE_TYPE_SETTING.getKey(), SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOT_STORE_TYPE);
- } else {
- expectedIndices.add(indexName);
- }
+ Metadata metadata = Metadata.builder().indices(nameToIndexMetadata).build();
+ ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metadata(metadata).build();
 
- Settings.Builder settingsBuilder = settings;
- IndexMetadata oldIndexMetadata = IndexMetadata.builder(indexName)
- .settings(settingsBuilder)
- .numberOfShards(1)
- .numberOfReplicas(0)
- .build();
- allIndices.add(oldIndexMetadata.getIndex());
- nameToIndexMetadata.put(oldIndexMetadata.getIndex().getName(), oldIndexMetadata);
- }
+ DeprecationIssue expected = new DeprecationIssue(
+ DeprecationIssue.Level.CRITICAL,
+ "Old data stream with a compatibility version < 9.0",
+ "https://www.elastic.co/guide/en/elasticsearch/reference/master/breaking-changes-9.0.html",
+ "This data stream has backing indices that were created before Elasticsearch 9.0.0",
+ false,
+ ofEntries(
+ entry("reindex_required", true),
+ entry("total_backing_indices", oldIndexCount + newIndexCount),
+ entry("indices_requiring_upgrade_count", expectedIndices.size()),
+ entry("indices_requiring_upgrade", expectedIndices)
+ )
+ );
 
- for (int i = 0; i < newIndexCount; i++) {
- Settings.Builder settingsBuilder = settings(IndexVersion.current());
- IndexMetadata newIndexMetadata = IndexMetadata.builder("new-data-stream-index-" + i)
- .settings(settingsBuilder)
- .numberOfShards(1)
- .numberOfReplicas(0)
- .build();
- allIndices.add(newIndexMetadata.getIndex());
- nameToIndexMetadata.put(newIndexMetadata.getIndex().getName(), newIndexMetadata);
- }
+ List<DeprecationIssue> issues = DeprecationChecks.filterChecks(DATA_STREAM_CHECKS, c -> c.apply(dataStream, clusterState));
 
- DataStream dataStream = new DataStream(
- randomAlphaOfLength(10),
- allIndices,
- randomNegativeLong(),
- Map.of(),
- randomBoolean(),
- false,
- false,
- randomBoolean(),
- randomFrom(IndexMode.values()),
- null,
- randomFrom(DataStreamOptions.EMPTY, DataStreamOptions.FAILURE_STORE_DISABLED, DataStreamOptions.FAILURE_STORE_ENABLED, null),
- List.of(),
- randomBoolean(),
- null
+ assertThat(issues, equalTo(singletonList(expected)));
+ }
+
+ public void testOldIndicesCheckWithOnlyClosedOrNewIndices() {
+ // This tests what happens when any old indices that we have are closed. We expect no deprecation warning.
+ int oldClosedIndexCount = randomIntBetween(1, 100);
+ int newOpenIndexCount = randomIntBetween(0, 100);
+ int newClosedIndexCount = randomIntBetween(0, 100);
+
+ Map<String, IndexMetadata> nameToIndexMetadata = new HashMap<>();
+ Set<String> expectedIndices = new HashSet<>();
+
+ DataStream dataStream = createTestDataStream(
+ 0,
+ oldClosedIndexCount,
+ newOpenIndexCount,
+ newClosedIndexCount,
+ nameToIndexMetadata,
+ expectedIndices
+ );
+
+ Metadata metadata = Metadata.builder().indices(nameToIndexMetadata).build();
+ ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metadata(metadata).build();
+
+ List<DeprecationIssue> issues = DeprecationChecks.filterChecks(DATA_STREAM_CHECKS, c -> c.apply(dataStream, clusterState));
+
+ assertThat(issues.size(), equalTo(0));
+ }
+
+ public void testOldIndicesCheckWithClosedAndOpenIndices() {
+ /*
+ * This tests what happens when a data stream has old indices, and some are open and some are closed. We expect a deprecation
+ * warning that includes information about the old ones only.
+ */
+ int oldOpenIndexCount = randomIntBetween(1, 100);
+ int oldClosedIndexCount = randomIntBetween(1, 100);
+ int newOpenIndexCount = randomIntBetween(0, 100);
+ int newClosedIndexCount = randomIntBetween(0, 100);
+
+ Map<String, IndexMetadata> nameToIndexMetadata = new HashMap<>();
+ Set<String> expectedIndices = new HashSet<>();
+
+ DataStream dataStream = createTestDataStream(
+ oldOpenIndexCount,
+ oldClosedIndexCount,
+ newOpenIndexCount,
+ newClosedIndexCount,
+ nameToIndexMetadata,
+ expectedIndices
  );
 
  Metadata metadata = Metadata.builder().indices(nameToIndexMetadata).build();
@@ -104,7 +127,7 @@ public void testOldIndicesCheck() {
  false,
  ofEntries(
  entry("reindex_required", true),
- entry("total_backing_indices", oldIndexCount + newIndexCount),
+ entry("total_backing_indices", oldOpenIndexCount + oldClosedIndexCount + newOpenIndexCount + newClosedIndexCount),
  entry("indices_requiring_upgrade_count", expectedIndices.size()),
  entry("indices_requiring_upgrade", expectedIndices)
  )
@@ -115,4 +138,90 @@ public void testOldIndicesCheck() {
  assertThat(issues, equalTo(singletonList(expected)));
  }
 
+ /*
+ * This creates a test DataStream with the given counts. The nameToIndexMetadata Map and the expectedIndices Set are mutable collections
+ * that will be populated by this method.
+ */
+ private DataStream createTestDataStream(
+ int oldOpenIndexCount,
+ int oldClosedIndexCount,
+ int newOpenIndexCount,
+ int newClosedIndexCount,
+ Map<String, IndexMetadata> nameToIndexMetadata,
+ Set<String> expectedIndices
+ ) {
+ List<Index> allIndices = new ArrayList<>();
+
+ for (int i = 0; i < oldOpenIndexCount; i++) {
+ allIndices.add(createOldIndex(i, false, nameToIndexMetadata, expectedIndices));
+ }
+ for (int i = 0; i < oldClosedIndexCount; i++) {
+ allIndices.add(createOldIndex(i, true, nameToIndexMetadata, null));
+ }
+ for (int i = 0; i < newOpenIndexCount; i++) {
+ allIndices.add(createNewIndex(i, false, nameToIndexMetadata));
+ }
+ for (int i = 0; i < newClosedIndexCount; i++) {
+ allIndices.add(createNewIndex(i, true, nameToIndexMetadata));
+ }
+
+ DataStream dataStream = new DataStream(
+ randomAlphaOfLength(10),
+ allIndices,
+ randomNegativeLong(),
+ Map.of(),
+ randomBoolean(),
+ false,
+ false,
+ randomBoolean(),
+ randomFrom(IndexMode.values()),
+ null,
+ randomFrom(DataStreamOptions.EMPTY, DataStreamOptions.FAILURE_STORE_DISABLED, DataStreamOptions.FAILURE_STORE_ENABLED, null),
+ List.of(),
+ randomBoolean(),
+ null
+ );
+ return dataStream;
+ }
+
+ private Index createOldIndex(
+ int suffix,
+ boolean isClosed,
+ Map<String, IndexMetadata> nameToIndexMetadata,
+ Set<String> expectedIndices
+ ) {
+ return createIndex(true, suffix, isClosed, nameToIndexMetadata, expectedIndices);
+ }
+
+ private Index createNewIndex(int suffix, boolean isClosed, Map<String, IndexMetadata> nameToIndexMetadata) {
+ return createIndex(false, suffix, isClosed, nameToIndexMetadata, null);
+ }
+
+ private Index createIndex(
+ boolean isOld,
+ int suffix,
+ boolean isClosed,
+ Map<String, IndexMetadata> nameToIndexMetadata,
+ Set<String> expectedIndices
+ ) {
+ Settings.Builder settingsBuilder = isOld ? settings(IndexVersion.fromId(7170099)) : settings(IndexVersion.current());
+ String indexName = (isOld ? "old-" : "new-") + (isClosed ? "closed-" : "") + "data-stream-index-" + suffix;
+ if (isOld && isClosed == false) { // we only expect warnings on open old indices
+ if (expectedIndices.isEmpty() == false && randomIntBetween(0, 2) == 0) {
+ settingsBuilder.put(INDEX_STORE_TYPE_SETTING.getKey(), SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOT_STORE_TYPE);
+ } else {
+ expectedIndices.add(indexName);
+ }
+ }
+ IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(indexName)
+ .settings(settingsBuilder)
+ .numberOfShards(1)
+ .numberOfReplicas(0);
+ if (isClosed) {
+ indexMetadataBuilder.state(IndexMetadata.State.CLOSE);
+ }
+ IndexMetadata indexMetadata = indexMetadataBuilder.build();
+ nameToIndexMetadata.put(indexMetadata.getIndex().getName(), indexMetadata);
+ return indexMetadata.getIndex();
+ }
 }

x-pack/plugin/deprecation/src/test/java/org/elasticsearch/xpack/deprecation/IndexDeprecationChecksTests.java

@@ -116,6 +116,22 @@ public void testOldIndicesCheckSnapshotIgnored() {
  assertThat(issues, empty());
  }
 
+ public void testOldIndicesCheckClosedIgnored() {
+ IndexVersion createdWith = IndexVersion.fromId(7170099);
+ Settings.Builder settings = settings(createdWith);
+ IndexMetadata indexMetadata = IndexMetadata.builder("test")
+ .settings(settings)
+ .numberOfShards(1)
+ .numberOfReplicas(0)
+ .state(IndexMetadata.State.CLOSE)
+ .build();
+ ClusterState clusterState = ClusterState.builder(ClusterState.EMPTY_STATE)
+ .metadata(Metadata.builder().put(indexMetadata, true))
+ .build();
+ List<DeprecationIssue> issues = DeprecationChecks.filterChecks(INDEX_SETTINGS_CHECKS, c -> c.apply(indexMetadata, clusterState));
+ assertThat(issues, empty());
+ }
+
  public void testTranslogRetentionSettings() {
  Settings.Builder settings = settings(IndexVersion.current());
  settings.put(IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getKey(), randomPositiveTimeValue());

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/DataStreamsUpgradeIT.java

@@ -24,6 +24,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -252,16 +253,27 @@ private static void createAndRolloverDataStream(String dataStreamName, int numRo
  assertOK(client().performRequest(putIndexTemplateRequest));
  bulkLoadData(dataStreamName);
  for (int i = 0; i < numRollovers; i++) {
- rollover(dataStreamName);
+ String oldIndexName = rollover(dataStreamName);
+ if (randomBoolean()) {
+ closeIndex(oldIndexName);
+ }
  bulkLoadData(dataStreamName);
  }
  }
 
  private void upgradeDataStream(String dataStreamName, int numRolloversOnOldCluster) throws Exception {
  Set<String> indicesNeedingUpgrade = getDataStreamIndices(dataStreamName);
+ Set<String> closedOldIndices = getClosedIndices(dataStreamName);
  final int explicitRolloverOnNewClusterCount = randomIntBetween(0, 2);
  for (int i = 0; i < explicitRolloverOnNewClusterCount; i++) {
- rollover(dataStreamName);
+ String oldIndexName = rollover(dataStreamName);
+ if (randomBoolean()) {
+ if (i == 0) {
+ // Since this is the first rollover on the new cluster, the old index came from the old cluster
+ closedOldIndices.add(oldIndexName);
+ }
+ closeIndex(oldIndexName);
+ }
  }
  Request reindexRequest = new Request("POST", "/_migration/reindex");
  reindexRequest.setJsonEntity(Strings.format("""
@@ -304,12 +316,14 @@ private void upgradeDataStream(String dataStreamName, int numRolloversOnOldClust
  */
  assertThat(
  statusResponseMap.get("total_indices_requiring_upgrade"),
- equalTo(originalWriteIndex + numRolloversOnOldCluster)
+ equalTo(originalWriteIndex + numRolloversOnOldCluster - closedOldIndices.size())
  );
- assertThat(statusResponseMap.get("successes"), equalTo(numRolloversOnOldCluster + 1));
+ assertThat(statusResponseMap.get("successes"), equalTo(numRolloversOnOldCluster + 1 - closedOldIndices.size()));
  // We expect all the original indices to have been deleted
  for (String oldIndex : indicesNeedingUpgrade) {
- assertThat(indexExists(oldIndex), equalTo(false));
+ if (closedOldIndices.contains(oldIndex) == false) {
+ assertThat(indexExists(oldIndex), equalTo(false));
+ }
  }
  assertThat(getDataStreamIndices(dataStreamName).size(), equalTo(expectedTotalIndicesInDataStream));
  }
@@ -329,6 +343,29 @@ private Set<String> getDataStreamIndices(String dataStreamName) throws IOExcepti
  return indices.stream().map(index -> index.get("index_name").toString()).collect(Collectors.toSet());
  }
 
+ @SuppressWarnings("unchecked")
+ private Set<String> getClosedIndices(String dataStreamName) throws IOException {
+ Set<String> allIndices = getDataStreamIndices(dataStreamName);
+ Set<String> closedIndices = new HashSet<>();
+ Response response = client().performRequest(new Request("GET", "_cluster/state/blocks/indices"));
+ Map<String, Object> responseMap = XContentHelper.convertToMap(JsonXContent.jsonXContent, response.getEntity().getContent(), false);
+ Map<String, Object> blocks = (Map<String, Object>) responseMap.get("blocks");
+ Map<String, Object> indices = (Map<String, Object>) blocks.get("indices");
+ for (Map.Entry<String, Object> indexEntry : indices.entrySet()) {
+ String indexName = indexEntry.getKey();
+ if (allIndices.contains(indexName)) {
+ Map<String, Object> blocksForIndex = (Map<String, Object>) indexEntry.getValue();
+ for (Map.Entry<String, Object> blockEntry : blocksForIndex.entrySet()) {
+ Map<String, String> block = (Map<String, String>) blockEntry.getValue();
+ if ("index closed".equals(block.get("description"))) {
+ closedIndices.add(indexName);
+ }
+ }
+ }
+ }
+ return closedIndices;
+ }
+
  /*
  * Similar to isOriginalClusterCurrent, but returns true if the major versions of the clusters are the same. So true
  * for 8.6 and 8.17, but false for 7.17 and 8.18.
@@ -370,9 +407,11 @@ static String formatInstant(Instant instant) {
  return DateFormatter.forPattern(FormatNames.STRICT_DATE_OPTIONAL_TIME.getName()).format(instant);
  }
 
- private static void rollover(String dataStreamName) throws IOException {
+ private static String rollover(String dataStreamName) throws IOException {
  Request rolloverRequest = new Request("POST", "/" + dataStreamName + "/_rollover");
  Response rolloverResponse = client().performRequest(rolloverRequest);
  assertOK(rolloverResponse);
+ String oldIndexName = (String) entityAsMap(rolloverResponse).get("old_index");
+ return oldIndexName;
  }
 }