Merge branch 'main' into remove-unused-subs

Author: shainaraskas

docs/changelog/133954.yaml

@@ -0,0 +1,6 @@
+pr: 133954
+summary: "ILM: Force merge on zero-replica cloned index before snapshotting for searchable snapshots"
+area: ILM+SLM
+type: enhancement
+issues:
+ - 75478

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/270_set_processor.yml

@@ -6,7 +6,7 @@ teardown:
         ignore: 404
 
 ---
-"Test set processor with template value":
+"Test set processor with a templated value":
   - do:
       ingest.put_pipeline:
         id: "1"
@@ -53,6 +53,7 @@ teardown:
         index: test
         id: "2"
   - match: { _source.foo: "hello" }
+
 ---
 "Test set processor with index change and require_alias":
   - do:
@@ -295,3 +296,38 @@ teardown:
   - match: { _source.result_2: "{{bar}}" }
   - match: { _source.old_bar: 2 }
   - match: { _source.bar: 3 }
+
+---
+"Test set processor with ignore_empty_value and a literal empty value":
+  - do:
+      ingest.put_pipeline:
+        id: "1"
+        body:  >
+          {
+            "processors": [
+              {
+                "set" : {
+                  "description": "we don't forbid this, even though it's a little silly",
+                  "field" : "foo",
+                  "value" : "",
+                  "ignore_empty_value" : true
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: "1"
+        pipeline: "1"
+        body: {
+          foo: "hello"
+        }
+
+  - do:
+      get:
+        index: test
+        id: "1"
+  - match: { _source.foo: "" }

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

@@ -40,7 +40,8 @@ public record LifecycleExecutionState(
     String snapshotName,
     String shrinkIndexName,
     String snapshotIndexName,
-    String downsampleIndexName
+    String downsampleIndexName,
+    String forceMergeCloneIndexName
 ) {
 
     public static final String ILM_CUSTOM_METADATA_KEY = "ilm";
@@ -64,6 +65,7 @@ public record LifecycleExecutionState(
     private static final String SNAPSHOT_INDEX_NAME = "snapshot_index_name";
     private static final String SHRINK_INDEX_NAME = "shrink_index_name";
     private static final String DOWNSAMPLE_INDEX_NAME = "rollup_index_name";
+    private static final String FORCE_MERGE_CLONE_INDEX_NAME = "force_merge_clone_index_name";
 
     public static final LifecycleExecutionState EMPTY_STATE = LifecycleExecutionState.builder().build();
 
@@ -89,7 +91,8 @@ public static Builder builder(LifecycleExecutionState state) {
             .setShrinkIndexName(state.shrinkIndexName)
             .setSnapshotIndexName(state.snapshotIndexName)
             .setDownsampleIndexName(state.downsampleIndexName)
-            .setStepTime(state.stepTime);
+            .setStepTime(state.stepTime)
+            .setForceMergeCloneIndexName(state.forceMergeCloneIndexName);
     }
 
     public static LifecycleExecutionState fromCustomMetadata(Map<String, String> customData) {
@@ -202,6 +205,10 @@ public static LifecycleExecutionState fromCustomMetadata(Map<String, String> cus
         if (downsampleIndexName != null) {
             builder.setDownsampleIndexName(downsampleIndexName);
         }
+        String forceMergeCloneIndexName = customData.get(FORCE_MERGE_CLONE_INDEX_NAME);
+        if (forceMergeCloneIndexName != null) {
+            builder.setForceMergeCloneIndexName(forceMergeCloneIndexName);
+        }
         return builder.build();
     }
 
@@ -274,6 +281,9 @@ public Map<String, String> asMap() {
         if (downsampleIndexName != null) {
             result.put(DOWNSAMPLE_INDEX_NAME, downsampleIndexName);
         }
+        if (forceMergeCloneIndexName != null) {
+            result.put(FORCE_MERGE_CLONE_INDEX_NAME, forceMergeCloneIndexName);
+        }
         return Collections.unmodifiableMap(result);
     }
 
@@ -307,6 +317,7 @@ public static class Builder {
         private String shrinkIndexName;
         private String snapshotIndexName;
         private String downsampleIndexName;
+        private String forceMergeCloneIndexName;
 
         public Builder setPhase(String phase) {
             this.phase = phase;
@@ -398,6 +409,11 @@ public Builder setDownsampleIndexName(String downsampleIndexName) {
             return this;
         }
 
+        public Builder setForceMergeCloneIndexName(String forceMergeCloneIndexName) {
+            this.forceMergeCloneIndexName = forceMergeCloneIndexName;
+            return this;
+        }
+
         public LifecycleExecutionState build() {
             return new LifecycleExecutionState(
                 phase,
@@ -417,7 +433,8 @@ public LifecycleExecutionState build() {
                 snapshotName,
                 shrinkIndexName,
                 snapshotIndexName,
-                downsampleIndexName
+                downsampleIndexName,
+                forceMergeCloneIndexName
             );
         }
     }

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -2087,6 +2087,10 @@ protected static void awaitIndexExists(String index, TimeValue timeout) throws I
         ensureHealth(client(), index, request -> request.addParameter("timeout", timeout.toString()));
     }
 
+    protected static void awaitIndexDoesNotExist(String index) throws Exception {
+        awaitIndexDoesNotExist(index, TimeValue.timeValueSeconds(10));
+    }
+
     protected static void awaitIndexDoesNotExist(String index, TimeValue timeout) throws Exception {
         assertBusy(() -> assertFalse(indexExists(index)), timeout.millis(), TimeUnit.MILLISECONDS);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/CreateSnapshotStep.java

@@ -101,8 +101,11 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
             );
             return;
         }
+        // If we performed the force merge step on the cloned index, we need to snapshot that index instead of the original.
+        final String clonedIndexName = lifecycleState.forceMergeCloneIndexName();
+        final String forceMergedIndexName = clonedIndexName != null ? clonedIndexName : indexName;
         CreateSnapshotRequest request = new CreateSnapshotRequest(TimeValue.MAX_VALUE, snapshotRepository, snapshotName);
-        request.indices(indexName);
+        request.indices(forceMergedIndexName);
         // this is safe as the snapshot creation will still be async, it's just that the listener will be notified when the snapshot is
         // complete
         request.waitForCompletion(true);
@@ -112,7 +115,7 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
             logger.debug(
                 "create snapshot response for policy [{}] and index [{}] is: {}",
                 policyName,
-                indexName,
+                forceMergedIndexName,
                 Strings.toString(response)
             );
             final SnapshotInfo snapInfo = response.getSnapshotInfo();
@@ -128,7 +131,7 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
                         snapshotRepository,
                         snapshotName,
                         policyName,
-                        indexName,
+                        forceMergedIndexName,
                         snapInfo.failedShards(),
                         snapInfo.totalShards()
                     )

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DeleteStep.java

@@ -16,26 +16,56 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.LifecycleExecutionState;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
 
+import java.util.function.BiFunction;
+
 /**
  * Deletes a single index.
  */
 public class DeleteStep extends AsyncRetryDuringSnapshotActionStep {
+
     public static final String NAME = "delete";
     private static final Logger logger = LogManager.getLogger(DeleteStep.class);
+    private static final BiFunction<String, LifecycleExecutionState, String> DEFAULT_TARGET_INDEX_NAME_SUPPLIER = (
+        indexName,
+        lifecycleState) -> indexName;
+
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
+    private final boolean indexSurvives;
 
+    /**
+     * Use this constructor to delete the index that ILM is currently operating on.
+     */
     public DeleteStep(StepKey key, StepKey nextStepKey, Client client) {
+        this(key, nextStepKey, client, DEFAULT_TARGET_INDEX_NAME_SUPPLIER, false);
+    }
+
+    /**
+     * Use this constructor to delete a specific index, potentially different from the one that ILM is currently operating on. The parameter
+     * {@code indexSurvives} indicates whether the index that ILM runs on will survive (i.e. not get deleted) this step.
+     * Look at the callers of {@link AsyncActionStep#indexSurvives()} for more details.
+     */
+    public DeleteStep(
+        StepKey key,
+        StepKey nextStepKey,
+        Client client,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier,
+        boolean indexSurvives
+    ) {
         super(key, nextStepKey, client);
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
+        this.indexSurvives = indexSurvives;
     }
 
     @Override
     public void performDuringNoSnapshot(IndexMetadata indexMetadata, ProjectMetadata currentProject, ActionListener<Void> listener) {
         String policyName = indexMetadata.getLifecyclePolicyName();
-        String indexName = indexMetadata.getIndex().getName();
+        String indexName = targetIndexNameSupplier.apply(indexMetadata.getIndex().getName(), indexMetadata.getLifecycleExecutionState());
         IndexAbstraction indexAbstraction = currentProject.getIndicesLookup().get(indexName);
         assert indexAbstraction != null : "invalid cluster metadata. index [" + indexName + "] was not found";
         DataStream dataStream = indexAbstraction.getParentDataStream();
@@ -88,7 +118,7 @@ public void performDuringNoSnapshot(IndexMetadata indexMetadata, ProjectMetadata
 
     @Override
     public boolean indexSurvives() {
-        return false;
+        return indexSurvives;
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ForceMergeStep.java

@@ -16,10 +16,12 @@
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.LifecycleExecutionState;
 import org.elasticsearch.common.Strings;
 
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.function.BiFunction;
 
 /**
  * Invokes a force merge on a single index.
@@ -28,11 +30,33 @@ public class ForceMergeStep extends AsyncActionStep {
 
     public static final String NAME = "forcemerge";
     private static final Logger logger = LogManager.getLogger(ForceMergeStep.class);
+    private static final BiFunction<String, LifecycleExecutionState, String> DEFAULT_TARGET_INDEX_NAME_SUPPLIER = (
+        indexName,
+        lifecycleState) -> indexName;
+
     private final int maxNumSegments;
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
 
+    /**
+     * Creates a new {@link ForceMergeStep} that will perform a force merge on the index that ILM is currently operating on.
+     */
     public ForceMergeStep(StepKey key, StepKey nextStepKey, Client client, int maxNumSegments) {
+        this(key, nextStepKey, client, maxNumSegments, DEFAULT_TARGET_INDEX_NAME_SUPPLIER);
+    }
+
+    /**
+     * Creates a new {@link ForceMergeStep} that will perform a force merge on the index name returned by the supplier.
+     */
+    public ForceMergeStep(
+        StepKey key,
+        StepKey nextStepKey,
+        Client client,
+        int maxNumSegments,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier
+    ) {
         super(key, nextStepKey, client);
         this.maxNumSegments = maxNumSegments;
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
     }
 
     @Override
@@ -51,7 +75,8 @@ public void performAction(
         ClusterStateObserver observer,
         ActionListener<Void> listener
     ) {
-        String indexName = indexMetadata.getIndex().getName();
+        String indexName = targetIndexNameSupplier.apply(indexMetadata.getIndex().getName(), indexMetadata.getLifecycleExecutionState());
+        assert indexName != null : "target index name supplier must not return null";
         ForceMergeRequest request = new ForceMergeRequest(indexName);
         request.maxNumSegments(maxNumSegments);
         getClient(currentState.projectId()).admin().indices().forceMerge(request, listener.delegateFailureAndWrap((l, response) -> {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/GenerateSnapshotNameStep.java

@@ -21,6 +21,7 @@
 
 import java.util.Locale;
 import java.util.Objects;
+import java.util.function.BiFunction;
 
 /**
  * Generates a snapshot name for the given index and records it in the index metadata along with the provided snapshot repository.
@@ -35,10 +36,17 @@ public class GenerateSnapshotNameStep extends ClusterStateActionStep {
     private static final Logger logger = LogManager.getLogger(GenerateSnapshotNameStep.class);
 
     private final String snapshotRepository;
-
-    public GenerateSnapshotNameStep(StepKey key, StepKey nextStepKey, String snapshotRepository) {
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
+
+    public GenerateSnapshotNameStep(
+        StepKey key,
+        StepKey nextStepKey,
+        String snapshotRepository,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier
+    ) {
         super(key, nextStepKey);
         this.snapshotRepository = snapshotRepository;
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
     }
 
     public String getSnapshotRepository() {
@@ -72,9 +80,11 @@ public ProjectState performAction(Index index, ProjectState projectState) {
                     + "] cannot continue until the repository is created or the policy is changed"
             );
         }
+        final String indexName = targetIndexNameSupplier.apply(index.getName(), lifecycleState);
+        assert indexName != null : "target index name supplier must not return null";
 
         LifecycleExecutionState.Builder newLifecycleState = LifecycleExecutionState.builder(lifecycleState);
-        newLifecycleState.setSnapshotIndexName(index.getName());
+        newLifecycleState.setSnapshotIndexName(indexName);
         newLifecycleState.setSnapshotRepository(snapshotRepository);
         if (lifecycleState.snapshotName() == null) {
             // generate and validate the snapshotName