Add support for partial searchable snapshots to ILM (elastic#68714)

This commit adds support for the recently introduced partial searchable snapshot (elastic#68509) to ILM.

Searchable snapshot ILM actions may now be specified with a `storage` option, specifying either
`full_copy` or `shared_cache` (similar to the "mount" API) to mount either a full or partial
searchable snapshot:

```json
PUT _ilm/policy/my_policy
{
  "policy": {
    "phases": {
      "cold": {
        "actions": {
          "searchable_snapshot" : {
            "snapshot_repository" : "backing_repo",
            "storage": "shared_cache"
          }
        }
      }
    }
  }
}
```

Internally, If more than one searchable snapshot action is specified (for example, a full searchable
snapshot in the "cold" phase and a partial searchable snapshot in the "frozen" phase) ILM will
re-use the existing snapshot when doing the second mount since a second snapshot is not required.

Currently this is allowed for actions that use the same repository, however, multiple
`searchable_snapshot` actions for the same index that use different repositories is not allowed (the
ERROR state is entered). We plan to allow this in the future in subsequent work.

If the `storage` option is not specified in the `searchable_snapshot` action, the mount type
defaults to "shared_cache" in the frozen phase and "full_copy" in all other phases.

Relates to elastic#68605

docs/reference/ilm/actions/ilm-searchable-snapshot.asciidoc

@@ -54,6 +54,12 @@ the shards are relocating, in which case they will not be merged.
 The `searchable_snapshot` action will continue executing even if not all shards
 are force merged.
 
+`storage`::
+(Optional, string)
+Specifies the type of snapshot that should be mounted for a searchable snapshot. This corresponds to
+the <<searchable-snapshots-api-mount-query-params, `storage` option when mounting a snapshot>>.
+Defaults to `full_copy` in non-frozen phases, or `shared_cache` in the frozen phase.
+
 [[ilm-searchable-snapshot-ex]]
 ==== Examples
 [source,console]
@@ -65,7 +71,8 @@ PUT _ilm/policy/my_policy
       "cold": {
         "actions": {
           "searchable_snapshot" : {
-            "snapshot_repository" : "backing_repo"
+            "snapshot_repository" : "backing_repo",
+            "storage": "shared_cache"
           }
         }
       }

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -1263,6 +1263,18 @@ public <E extends Enum<E>> E readEnum(Class<E> enumClass) throws IOException {
         return readEnum(enumClass, enumClass.getEnumConstants());
     }
 
+    /**
+     * Reads an optional enum with type E that was serialized based on the value of its ordinal
+     */
+    @Nullable
+    public <E extends Enum<E>> E readOptionalEnum(Class<E> enumClass) throws IOException {
+        if (readBoolean()) {
+            return readEnum(enumClass, enumClass.getEnumConstants());
+        } else {
+            return null;
+        }
+    }
+
     private <E extends Enum<E>> E readEnum(Class<E> enumClass, E[] values) throws IOException {
         int ordinal = readVInt();
         if (ordinal < 0 || ordinal >= values.length) {

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -1245,6 +1245,18 @@ public <E extends Enum<E>> void writeEnum(E enumValue) throws IOException {
         writeVInt(enumValue.ordinal());
     }
 
+    /**
+     * Writes an optional enum with type E based on its ordinal value
+     */
+    public <E extends Enum<E>> void writeOptionalEnum(@Nullable E enumValue) throws IOException {
+        if (enumValue == null) {
+            writeBoolean(false);
+        } else {
+            writeBoolean(true);
+            writeVInt(enumValue.ordinal());
+        }
+    }
+
     /**
      * Writes an EnumSet with type E that by serialized it based on it's ordinal value
      */

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

@@ -85,6 +85,7 @@ public ClusterState performAction(Index index, ClusterState clusterState) {
         }
         relevantTargetCustomData.setSnapshotRepository(lifecycleState.getSnapshotRepository());
         relevantTargetCustomData.setSnapshotName(lifecycleState.getSnapshotName());
+        relevantTargetCustomData.setSnapshotIndexName(lifecycleState.getSnapshotIndexName());
 
         Metadata.Builder newMetadata = Metadata.builder(clusterState.getMetadata())
             .put(IndexMetadata.builder(targetIndexMetadata)

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

@@ -79,6 +79,7 @@ public ClusterState performAction(Index index, ClusterState clusterState) {
         }
         newCustomData.setSnapshotName(snapshotName);
         newCustomData.setSnapshotRepository(snapshotRepository);
+        newCustomData.setSnapshotIndexName(index.getName());
 
         IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(indexMetaData);
         indexMetadataBuilder.putCustom(ILM_CUSTOM_METADATA_KEY, newCustomData.build().asMap());

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

@@ -37,8 +37,9 @@ public class LifecycleExecutionState {
     private static final String FAILED_STEP_RETRY_COUNT = "failed_step_retry_count";
     private static final String STEP_INFO = "step_info";
     private static final String PHASE_DEFINITION = "phase_definition";
-    private static final String SNAPSHOT_NAME ="snapshot_name";
-    private static final String SNAPSHOT_REPOSITORY ="snapshot_repository";
+    private static final String SNAPSHOT_NAME = "snapshot_name";
+    private static final String SNAPSHOT_REPOSITORY = "snapshot_repository";
+    private static final String SNAPSHOT_INDEX_NAME = "snapshot_index_name";
 
     private final String phase;
     private final String action;
@@ -54,10 +55,12 @@ public class LifecycleExecutionState {
     private final Long stepTime;
     private final String snapshotName;
     private final String snapshotRepository;
+    private final String snapshotIndexName;
 
     private LifecycleExecutionState(String phase, String action, String step, String failedStep, Boolean isAutoRetryableError,
                                     Integer failedStepRetryCount, String stepInfo, String phaseDefinition, Long lifecycleDate,
-                                    Long phaseTime, Long actionTime, Long stepTime, String snapshotRepository, String snapshotName) {
+                                    Long phaseTime, Long actionTime, Long stepTime, String snapshotRepository, String snapshotName,
+                                    String snapshotIndexName) {
         this.phase = phase;
         this.action = action;
         this.step = step;
@@ -72,6 +75,7 @@ private LifecycleExecutionState(String phase, String action, String step, String
         this.stepTime = stepTime;
         this.snapshotRepository = snapshotRepository;
         this.snapshotName = snapshotName;
+        this.snapshotIndexName = snapshotIndexName;
     }
 
     /**
@@ -131,6 +135,7 @@ public static Builder builder(LifecycleExecutionState state) {
             .setActionTime(state.actionTime)
             .setSnapshotRepository(state.snapshotRepository)
             .setSnapshotName(state.snapshotName)
+            .setSnapshotIndexName(state.snapshotIndexName)
             .setStepTime(state.stepTime);
     }
 
@@ -198,6 +203,9 @@ static LifecycleExecutionState fromCustomMetadata(Map<String, String> customData
                     e, STEP_TIME, customData.get(STEP_TIME));
             }
         }
+        if (customData.containsKey(SNAPSHOT_INDEX_NAME)) {
+            builder.setSnapshotIndexName(customData.get(SNAPSHOT_INDEX_NAME));
+        }
         return builder.build();
     }
 
@@ -250,6 +258,9 @@ public Map<String, String> asMap() {
         if (snapshotName != null) {
             result.put(SNAPSHOT_NAME, snapshotName);
         }
+        if (snapshotIndexName != null) {
+            result.put(SNAPSHOT_INDEX_NAME, snapshotIndexName);
+        }
         return Collections.unmodifiableMap(result);
     }
 
@@ -309,6 +320,10 @@ public String getSnapshotRepository() {
         return snapshotRepository;
     }
 
+    public String getSnapshotIndexName() {
+        return snapshotIndexName;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -327,14 +342,15 @@ public boolean equals(Object o) {
             Objects.equals(getStepInfo(), that.getStepInfo()) &&
             Objects.equals(getSnapshotRepository(), that.getSnapshotRepository()) &&
             Objects.equals(getSnapshotName(), that.getSnapshotName()) &&
+            Objects.equals(getSnapshotIndexName(), that.getSnapshotIndexName()) &&
             Objects.equals(getPhaseDefinition(), that.getPhaseDefinition());
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(getPhase(), getAction(), getStep(), getFailedStep(), isAutoRetryableError(), getFailedStepRetryCount(),
             getStepInfo(), getPhaseDefinition(), getLifecycleDate(), getPhaseTime(), getActionTime(), getStepTime(),
-            getSnapshotRepository(), getSnapshotName());
+            getSnapshotRepository(), getSnapshotName(), getSnapshotIndexName());
     }
 
     @Override
@@ -357,6 +373,7 @@ public static class Builder {
         private Integer failedStepRetryCount;
         private String snapshotName;
         private String snapshotRepository;
+        private String snapshotIndexName;
 
         public Builder setPhase(String phase) {
             this.phase = phase;
@@ -428,9 +445,14 @@ public Builder setSnapshotName(String snapshotName) {
             return this;
         }
 
+        public Builder setSnapshotIndexName(String snapshotIndexName) {
+            this.snapshotIndexName = snapshotIndexName;
+            return this;
+        }
+
         public LifecycleExecutionState build() {
             return new LifecycleExecutionState(phase, action, step, failedStep, isAutoRetryableError, failedStepRetryCount, stepInfo,
-                phaseDefinition, indexCreationDate, phaseTime, actionTime, stepTime, snapshotRepository, snapshotName);
+                phaseDefinition, indexCreationDate, phaseTime, actionTime, stepTime, snapshotRepository, snapshotName, snapshotIndexName);
         }
     }
 

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

@@ -34,10 +34,13 @@ public class MountSnapshotStep extends AsyncRetryDuringSnapshotActionStep {
     private static final Logger logger = LogManager.getLogger(MountSnapshotStep.class);
 
     private final String restoredIndexPrefix;
+    private final MountSearchableSnapshotRequest.Storage storageType;
 
-    public MountSnapshotStep(StepKey key, StepKey nextStepKey, Client client, String restoredIndexPrefix) {
+    public MountSnapshotStep(StepKey key, StepKey nextStepKey, Client client, String restoredIndexPrefix,
+                             MountSearchableSnapshotRequest.Storage storageType) {
         super(key, nextStepKey, client);
         this.restoredIndexPrefix = restoredIndexPrefix;
+        this.storageType = Objects.requireNonNull(storageType, "a storage type must be specified");
     }
 
     @Override
@@ -49,9 +52,13 @@ public String getRestoredIndexPrefix() {
         return restoredIndexPrefix;
     }
 
+    public MountSearchableSnapshotRequest.Storage getStorage() {
+        return storageType;
+    }
+
     @Override
     void performDuringNoSnapshot(IndexMetadata indexMetadata, ClusterState currentClusterState, Listener listener) {
-        final String indexName = indexMetadata.getIndex().getName();
+        String indexName = indexMetadata.getIndex().getName();
 
         LifecycleExecutionState lifecycleState = fromIndexMetadata(indexMetadata);
 
@@ -71,13 +78,29 @@ void performDuringNoSnapshot(IndexMetadata indexMetadata, ClusterState currentCl
         }
 
         String mountedIndexName = restoredIndexPrefix + indexName;
-        if(currentClusterState.metadata().index(mountedIndexName) != null) {
+        if (currentClusterState.metadata().index(mountedIndexName) != null) {
             logger.debug("mounted index [{}] for policy [{}] and index [{}] already exists. will not attempt to mount the index again",
                 mountedIndexName, policyName, indexName);
             listener.onResponse(true);
             return;
         }
 
+        final String snapshotIndexName = lifecycleState.getSnapshotIndexName();
+        if (snapshotIndexName == null) {
+            // This index had its searchable snapshot created prior to a version where we captured
+            // the original index name, so make our best guess at the name
+            indexName = bestEffortIndexNameResolution(indexName);
+            logger.debug("index [{}] using policy [{}] does not have a stored snapshot index name, " +
+                "using our best effort guess of [{}] for the original snapshotted index name",
+                indexMetadata.getIndex().getName(), policyName, indexName);
+        } else {
+            // Use the name of the snapshot as specified in the metadata, because the current index
+            // name not might not reflect the name of the index actually in the snapshot
+            logger.debug("index [{}] using policy [{}] has a different name [{}] within the snapshot to be restored, " +
+                "using the snapshot index name from generated metadata for mounting", indexName, policyName, snapshotIndexName);
+            indexName = snapshotIndexName;
+        }
+
         final MountSearchableSnapshotRequest mountSearchableSnapshotRequest = new MountSearchableSnapshotRequest(mountedIndexName,
             snapshotRepository, snapshotName, indexName, Settings.builder()
             .put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), Boolean.FALSE.toString())
@@ -91,8 +114,7 @@ void performDuringNoSnapshot(IndexMetadata indexMetadata, ClusterState currentCl
             // we'll not wait for the snapshot to complete in this step as the async steps are executed from threads that shouldn't
             // perform expensive operations (ie. clusterStateProcessed)
             false,
-            // restoring into the cold tier, so use a full local copy
-            MountSearchableSnapshotRequest.Storage.FULL_COPY);
+            storageType);
         getClient().execute(MountSearchableSnapshotAction.INSTANCE, mountSearchableSnapshotRequest,
             ActionListener.wrap(response -> {
                 if (response.status() != RestStatus.OK && response.status() != RestStatus.ACCEPTED) {
@@ -103,9 +125,21 @@ void performDuringNoSnapshot(IndexMetadata indexMetadata, ClusterState currentCl
             }, listener::onFailure));
     }
 
+    /**
+     * Tries to guess the original index name given the current index name, tries to drop the
+     * "partial-" and "restored-" prefixes, since those are what ILM uses. Does not handle
+     * unorthodox cases like "restored-partial-[indexname]" since this is not intended to be
+     * exhaustive.
+     */
+    static String bestEffortIndexNameResolution(String indexName) {
+        String originalName = indexName.replaceFirst("^" + SearchableSnapshotAction.PARTIAL_RESTORED_INDEX_PREFIX, "");
+        originalName = originalName.replaceFirst("^" + SearchableSnapshotAction.FULL_RESTORED_INDEX_PREFIX, "");
+        return originalName;
+    }
+
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), restoredIndexPrefix);
+        return Objects.hash(super.hashCode(), restoredIndexPrefix, storageType);
     }
 
     @Override
@@ -117,6 +151,8 @@ public boolean equals(Object obj) {
             return false;
         }
         MountSnapshotStep other = (MountSnapshotStep) obj;
-        return super.equals(obj) && Objects.equals(restoredIndexPrefix, other.restoredIndexPrefix);
+        return super.equals(obj) &&
+            Objects.equals(restoredIndexPrefix, other.restoredIndexPrefix) &&
+            Objects.equals(storageType, other.storageType);
     }
 }