Handling frozen indices in reindex data streams (#122279)

masseyke · web-flow · commit 0f6ab6c1b97a · 2025-02-12T18:57:24.000+01:00
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/InternalUsers.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/InternalUsers.java
@@ -29,6 +29,7 @@
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.index.reindex.ReindexAction;
 import org.elasticsearch.xpack.core.XPackPlugin;
+import org.elasticsearch.xpack.core.frozen.action.FreezeIndexAction;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.support.MetadataUtils;
 
@@ -215,6 +216,7 @@ public class InternalUsers {
                         TransportUpdateSettingsAction.TYPE.name(),
                         RefreshAction.NAME,
                         ReindexAction.NAME,
+                        FreezeIndexAction.NAME,
                         TransportSearchAction.NAME,
                         TransportBulkAction.NAME,
                         TransportIndexAction.NAME,
diff --git a/x-pack/plugin/migrate/build.gradle b/x-pack/plugin/migrate/build.gradle
@@ -17,6 +17,7 @@ dependencies {
   compileOnly project(path: xpackModule('core'))
   testImplementation(testArtifact(project(xpackModule('core'))))
   testImplementation project(xpackModule('ccr'))
+  testImplementation project(xpackModule('frozen-indices'))
   testImplementation project(':modules:data-streams')
   testImplementation project(path: ':modules:reindex')
   testImplementation project(path: ':modules:ingest-common')
diff --git a/x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java b/x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java
@@ -9,6 +9,8 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
@@ -38,25 +40,34 @@
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.compress.CompressedXContent;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.FormatNames;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.datastreams.DataStreamsPlugin;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.frozen.FrozenEngine;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.ingest.common.IngestCommonPlugin;
+import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.protocol.xpack.frozen.FreezeRequest;
+import org.elasticsearch.protocol.xpack.frozen.FreezeResponse;
 import org.elasticsearch.reindex.ReindexPlugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.core.frozen.action.FreezeIndexAction;
+import org.elasticsearch.xpack.frozen.action.TransportFreezeIndexAction;
 import org.elasticsearch.xpack.migrate.MigratePlugin;
 import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
 import org.junit.Before;
 
 import java.io.IOException;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
@@ -69,6 +80,7 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
 
 public class ReindexDatastreamIndexTransportActionIT extends ESIntegTestCase {
 
@@ -112,7 +124,8 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
             ReindexPlugin.class,
             MockTransportService.TestPlugin.class,
             DataStreamsPlugin.class,
-            IngestCommonPlugin.class
+            IngestCommonPlugin.class,
+            TestFrozenIndicesPlugin.class
         );
     }
 
@@ -595,6 +608,43 @@ public void testTsdbStartEndSet() throws Exception {
         assertHitCount(prepareSearch(destIndex).setSize(0), 1);
     }
 
+    public void testIndexUnfrozen() {
+        var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
+        safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));
+
+        // add doc with timestamp
+        String time = DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.formatMillis(System.currentTimeMillis());
+        var doc = String.format(Locale.ROOT, "{\"%s\":\"%s\"}", DEFAULT_TIMESTAMP_FIELD, time);
+        addDoc(sourceIndex, doc);
+        FreezeRequest freezeRequest = new FreezeRequest(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, sourceIndex);
+        freezeRequest.setFreeze(true);
+        FreezeResponse freezeResponse = safeGet(client().execute(FreezeIndexAction.INSTANCE, freezeRequest));
+        assertAcked(freezeResponse);
+        assertThat(
+            safeGet(admin().indices().getSettings(new GetSettingsRequest().indices(sourceIndex))).getIndexToSettings()
+                .get(sourceIndex)
+                .get(FrozenEngine.INDEX_FROZEN.getKey()),
+            not(equalTo(null))
+        );
+
+        String destIndex = safeGet(
+            client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
+        ).getDestIndex();
+
+        assertThat(
+            safeGet(admin().indices().getSettings(new GetSettingsRequest().indices(sourceIndex))).getIndexToSettings()
+                .get(sourceIndex)
+                .get(FrozenEngine.INDEX_FROZEN.getKey()),
+            equalTo(null)
+        );
+        assertThat(
+            safeGet(admin().indices().getSettings(new GetSettingsRequest().indices(destIndex))).getIndexToSettings()
+                .get(destIndex)
+                .get(FrozenEngine.INDEX_FROZEN.getKey()),
+            equalTo(null)
+        );
+    }
+
     private static void cleanupMetadataBlocks(String index) {
         var settings = Settings.builder()
             .putNull(IndexMetadata.SETTING_READ_ONLY)
@@ -635,4 +685,22 @@ private void ensureHealth(String index) {
             ensureYellow(index);
         }
     }
+
+    /*
+     * This takes the place of the real FrozenIndices plugin. We can't use that one because its EngineFactory conflicts with the one used
+     * by this test. We only need the settings and the mapping of the FreezeIndexAction though.
+     */
+    public static class TestFrozenIndicesPlugin extends Plugin implements ActionPlugin {
+        @Override
+        public List<Setting<?>> getSettings() {
+            return Arrays.asList(FrozenEngine.INDEX_FROZEN);
+        }
+
+        @Override
+        public List<ActionPlugin.ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+            List<ActionPlugin.ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>();
+            actions.add(new ActionPlugin.ActionHandler<>(FreezeIndexAction.INSTANCE, TransportFreezeIndexAction.class));
+            return actions;
+        }
+    }
 }
diff --git a/x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java b/x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java
@@ -42,17 +42,21 @@
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.frozen.FrozenEngine;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.ReindexAction;
 import org.elasticsearch.index.reindex.ReindexRequest;
 import org.elasticsearch.index.reindex.ScrollableHitSource;
 import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.protocol.xpack.frozen.FreezeRequest;
+import org.elasticsearch.protocol.xpack.frozen.FreezeResponse;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.deprecation.DeprecatedIndexPredicate;
+import org.elasticsearch.xpack.core.frozen.action.FreezeIndexAction;
 import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
 
 import java.util.Locale;
@@ -156,7 +160,8 @@ protected void doExecute(
             return;
         }
         final boolean wasClosed = isClosed(sourceIndex);
-        SubscribableListener.<AcknowledgedResponse>newForked(l -> setBlockWrites(sourceIndexName, l, taskId))
+        SubscribableListener.<FreezeResponse>newForked(l -> unfreezeIfFrozen(sourceIndexName, sourceIndex, l, taskId))
+            .<AcknowledgedResponse>andThen(l -> setBlockWrites(sourceIndexName, l, taskId))
             .<OpenIndexResponse>andThen(l -> openIndexIfClosed(sourceIndexName, wasClosed, l, taskId))
             .<BroadcastResponse>andThen(l -> refresh(sourceIndexName, l, taskId))
             .<AcknowledgedResponse>andThen(l -> deleteDestIfExists(destIndexName, l, taskId))
@@ -200,6 +205,22 @@ private static boolean isClosed(IndexMetadata indexMetadata) {
         return indexMetadata.getState().equals(IndexMetadata.State.CLOSE);
     }
 
+    private void unfreezeIfFrozen(
+        String sourceIndexName,
+        IndexMetadata indexMetadata,
+        ActionListener<FreezeResponse> listener,
+        TaskId parentTaskId
+    ) {
+        if (FrozenEngine.INDEX_FROZEN.get(indexMetadata.getSettings()).equals(Boolean.TRUE)) {
+            logger.debug("Unfreezing source index [{}]", sourceIndexName);
+            FreezeRequest freezeRequest = new FreezeRequest(TimeValue.MAX_VALUE, TimeValue.MAX_VALUE, sourceIndexName).setFreeze(false);
+            freezeRequest.setParentTask(parentTaskId);
+            client.execute(FreezeIndexAction.INSTANCE, freezeRequest, listener);
+        } else {
+            listener.onResponse(null);
+        }
+    }
+
     private void setBlockWrites(String sourceIndexName, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
         logger.debug("Setting write block on source index [{}]", sourceIndexName);
         addBlockToIndex(WRITE, sourceIndexName, new ActionListener<>() {
diff --git a/x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/DataStreamsUpgradeIT.java b/x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/DataStreamsUpgradeIT.java
@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.upgrades;
 
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.Build;
 import org.elasticsearch.TransportVersions;
@@ -342,12 +343,29 @@ private void createAndRolloverDataStream(String dataStreamName, int numRollovers
         for (int i = 0; i < numRollovers; i++) {
             String oldIndexName = rollover(dataStreamName);
             if (randomBoolean()) {
-                closeIndex(oldIndexName);
+                closeOrFreezeIndex(oldIndexName);
             }
             bulkLoadData(dataStreamName);
         }
     }
 
+    // Randomly either closes or freezes the index. If the cluster does not support the _freeze API, then this always closes the index.
+    private void closeOrFreezeIndex(String indexName) throws IOException {
+        boolean canFreeze = minimumTransportVersion().before(TransportVersions.V_8_0_0);
+        if (canFreeze && randomBoolean()) {
+            final Request freezeRequest = new Request(HttpPost.METHOD_NAME, "/" + indexName + "/_freeze");
+            freezeRequest.setOptions(
+                expectWarnings(
+                    "Frozen indices are deprecated because they provide no benefit given improvements in heap memory utilization. "
+                        + "They will be removed in a future release."
+                )
+            );
+            assertOK(client().performRequest(freezeRequest));
+        } else {
+            closeIndex(indexName);
+        }
+    }
+
     private static void createIlmPolicy() throws IOException {
         String ilmPolicy = """
             {
