Refactor remote store flow to support any path type (#12822)

Signed-off-by: Ashish Singh <ssashish@amazon.com>

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java

@@ -88,8 +88,8 @@
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.mapper.MapperService.MergeReason;
 import org.opensearch.index.query.QueryShardContext;
-import org.opensearch.index.remote.RemoteStorePathResolver;
 import org.opensearch.index.remote.RemoteStorePathType;
+import org.opensearch.index.remote.RemoteStorePathTypeResolver;
 import org.opensearch.index.shard.IndexSettingProvider;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.IndexCreationException;
@@ -113,6 +113,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -170,7 +171,7 @@ public class MetadataCreateIndexService {
     private AwarenessReplicaBalance awarenessReplicaBalance;
 
     @Nullable
-    private final RemoteStorePathResolver remoteStorePathResolver;
+    private final RemoteStorePathTypeResolver remoteStorePathTypeResolver;
 
     public MetadataCreateIndexService(
         final Settings settings,
@@ -203,8 +204,8 @@ public MetadataCreateIndexService(
 
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
         createIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CREATE_INDEX_KEY, true);
-        remoteStorePathResolver = isRemoteDataAttributePresent(settings)
-            ? new RemoteStorePathResolver(clusterService.getClusterSettings())
+        remoteStorePathTypeResolver = isRemoteDataAttributePresent(settings)
+            ? new RemoteStorePathTypeResolver(clusterService.getClusterSettings())
             : null;
     }
 
@@ -553,7 +554,7 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
         tmpImdBuilder.setRoutingNumShards(routingNumShards);
         tmpImdBuilder.settings(indexSettings);
         tmpImdBuilder.system(isSystem);
-        addRemoteCustomData(tmpImdBuilder);
+        addRemoteStorePathTypeInCustomData(tmpImdBuilder, true);
 
         // Set up everything, now locally create the index to see that things are ok, and apply
         IndexMetadata tempMetadata = tmpImdBuilder.build();
@@ -562,17 +563,24 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
         return tempMetadata;
     }
 
-    public void addRemoteCustomData(IndexMetadata.Builder tmpImdBuilder) {
-        if (remoteStorePathResolver != null) {
+    /**
+     * Adds the remote store path type information in custom data of index metadata.
+     *
+     * @param tmpImdBuilder     index metadata builder.
+     * @param assertNullOldType flag to verify that the old remote store path type is null
+     */
+    public void addRemoteStorePathTypeInCustomData(IndexMetadata.Builder tmpImdBuilder, boolean assertNullOldType) {
+        if (remoteStorePathTypeResolver != null) {
             // It is possible that remote custom data exists already. In such cases, we need to only update the path type
             // in the remote store custom data map.
             Map<String, String> existingRemoteCustomData = tmpImdBuilder.removeCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
             Map<String, String> remoteCustomData = existingRemoteCustomData == null
                 ? new HashMap<>()
                 : new HashMap<>(existingRemoteCustomData);
             // Determine the path type for use using the remoteStorePathResolver.
-            String newPathType = remoteStorePathResolver.resolveType().toString();
+            String newPathType = remoteStorePathTypeResolver.getType().toString();
             String oldPathType = remoteCustomData.put(RemoteStorePathType.NAME, newPathType);
+            assert !assertNullOldType || Objects.isNull(oldPathType);
             logger.trace(() -> new ParameterizedMessage("Added new path type {}, replaced old path type {}", newPathType, oldPathType));
             tmpImdBuilder.putCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY, remoteCustomData);
         }

server/src/main/java/org/opensearch/index/IndexService.java

@@ -507,7 +507,8 @@ public synchronized IndexShard createShard(
                     remoteDirectory = ((RemoteSegmentStoreDirectoryFactory) remoteDirectoryFactory).newDirectory(
                         RemoteStoreNodeAttribute.getRemoteStoreSegmentRepo(this.indexSettings.getNodeSettings()),
                         this.indexSettings.getUUID(),
-                        shardId
+                        shardId,
+                        this.indexSettings.getRemoteStorePathType()
                     );
                 }
                 remoteStore = new Store(shardId, this.indexSettings, remoteDirectory, lock, Store.OnClose.EMPTY, path);

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -48,6 +48,7 @@
 import org.opensearch.core.common.unit.ByteSizeUnit;
 import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.core.index.Index;
+import org.opensearch.index.remote.RemoteStorePathType;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.ingest.IngestService;
@@ -59,6 +60,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -1905,4 +1907,11 @@ public double getDocIdFuzzySetFalsePositiveProbability() {
     public void setDocIdFuzzySetFalsePositiveProbability(double docIdFuzzySetFalsePositiveProbability) {
         this.docIdFuzzySetFalsePositiveProbability = docIdFuzzySetFalsePositiveProbability;
     }
+
+    public RemoteStorePathType getRemoteStorePathType() {
+        Map<String, String> remoteCustomData = indexMetadata.getCustomData(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
+        return remoteCustomData != null && remoteCustomData.containsKey(RemoteStorePathType.NAME)
+            ? RemoteStorePathType.parseString(remoteCustomData.get(RemoteStorePathType.NAME))
+            : RemoteStorePathType.FIXED;
+    }
 }

server/src/main/java/org/opensearch/index/remote/RemoteStoreDataEnums.java

@@ -0,0 +1,65 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.remote;
+
+import java.util.Set;
+
+import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.DATA;
+import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.METADATA;
+
+/**
+ * This class contains the different enums related to remote store data categories and types.
+ *
+ * @opensearch.internal
+ */
+public class RemoteStoreDataEnums {
+
+    /**
+     * Categories of the data in Remote store.
+     */
+    public enum DataCategory {
+        SEGMENTS("segments", Set.of(DataType.values())),
+        TRANSLOG("translog", Set.of(DATA, METADATA));
+
+        private final String name;
+        private final Set<DataType> supportedDataTypes;
+
+        DataCategory(String name, Set<DataType> supportedDataTypes) {
+            this.name = name;
+            this.supportedDataTypes = supportedDataTypes;
+        }
+
+        public boolean isSupportedDataType(DataType dataType) {
+            return supportedDataTypes.contains(dataType);
+        }
+
+        public String getName() {
+            return name;
+        }
+    }
+
+    /**
+     * Types of data in remote store.
+     */
+    public enum DataType {
+        DATA("data"),
+        METADATA("metadata"),
+        LOCK_FILES("lock_files");
+
+        private final String name;
+
+        DataType(String name) {
+            this.name = name;
+        }
+
+        public String getName() {
+            return name;
+        }
+    }
+}

server/src/main/java/org/opensearch/index/remote/RemoteStorePathType.java

@@ -8,6 +8,10 @@
 
 package org.opensearch.index.remote;
 
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.index.remote.RemoteStoreDataEnums.DataCategory;
+import org.opensearch.index.remote.RemoteStoreDataEnums.DataType;
+
 import java.util.Locale;
 
 /**
@@ -18,13 +22,46 @@
  */
 public enum RemoteStorePathType {
 
-    FIXED,
-    HASHED_PREFIX;
+    FIXED {
+        @Override
+        public BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
+            return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
+        }
+    },
+    HASHED_PREFIX {
+        @Override
+        public BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
+            // TODO - We need to implement this, keeping the same path as Fixed for sake of multiple tests that can fail otherwise.
+            // throw new UnsupportedOperationException("Not implemented"); --> Not using this for unblocking couple of tests.
+            return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
+        }
+    };
+
+    /**
+     * @param basePath     base path of the underlying blob store repository
+     * @param indexUUID    of the index
+     * @param shardId      shard id
+     * @param dataCategory is either translog or segment
+     * @param dataType     can be one of data, metadata or lock_files.
+     * @return the blob path for the underlying remote store path type.
+     */
+    public BlobPath path(BlobPath basePath, String indexUUID, String shardId, DataCategory dataCategory, DataType dataType) {
+        assert dataCategory.isSupportedDataType(dataType) : "category:"
+            + dataCategory
+            + " type:"
+            + dataType
+            + " are not supported together";
+        return generatePath(basePath, indexUUID, shardId, dataCategory.getName(), dataType.getName());
+    }
+
+    abstract BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType);
 
     public static RemoteStorePathType parseString(String remoteStoreBlobPathType) {
         try {
             return RemoteStorePathType.valueOf(remoteStoreBlobPathType.toUpperCase(Locale.ROOT));
-        } catch (IllegalArgumentException e) {
+        } catch (IllegalArgumentException | NullPointerException e) {
+            // IllegalArgumentException is thrown when the input does not match any enum name
+            // NullPointerException is thrown when the input is null
             throw new IllegalArgumentException("Could not parse RemoteStorePathType for [" + remoteStoreBlobPathType + "]");
         }
     }

server/src/main/java/org/opensearch/index/remote/RemoteStorePathResolver.java → server/src/main/java/org/opensearch/index/remote/RemoteStorePathTypeResolver.java

@@ -16,15 +16,20 @@
  *
  * @opensearch.internal
  */
-public class RemoteStorePathResolver {
+public class RemoteStorePathTypeResolver {
 
-    private final ClusterSettings clusterSettings;
+    private volatile RemoteStorePathType type;
 
-    public RemoteStorePathResolver(ClusterSettings clusterSettings) {
-        this.clusterSettings = clusterSettings;
+    public RemoteStorePathTypeResolver(ClusterSettings clusterSettings) {
+        type = clusterSettings.get(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING);
+        clusterSettings.addSettingsUpdateConsumer(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING, this::setType);
     }
 
-    public RemoteStorePathType resolveType() {
-        return clusterSettings.get(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING);
+    public RemoteStorePathType getType() {
+        return type;
+    }
+
+    public void setType(RemoteStorePathType type) {
+        this.type = type;
     }
 }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -4932,7 +4932,7 @@ public void deleteTranslogFilesFromRemoteTranslog() throws IOException {
         TranslogFactory translogFactory = translogFactorySupplier.apply(indexSettings, shardRouting);
         assert translogFactory instanceof RemoteBlobStoreInternalTranslogFactory;
         Repository repository = ((RemoteBlobStoreInternalTranslogFactory) translogFactory).getRepository();
-        RemoteFsTranslog.cleanup(repository, shardId, getThreadPool());
+        RemoteFsTranslog.cleanup(repository, shardId, getThreadPool(), indexSettings.getRemoteStorePathType());
     }
 
     /*
@@ -4949,7 +4949,14 @@ public void syncTranslogFilesFromRemoteTranslog() throws IOException {
         TranslogFactory translogFactory = translogFactorySupplier.apply(indexSettings, shardRouting);
         assert translogFactory instanceof RemoteBlobStoreInternalTranslogFactory;
         Repository repository = ((RemoteBlobStoreInternalTranslogFactory) translogFactory).getRepository();
-        RemoteFsTranslog.download(repository, shardId, getThreadPool(), shardPath().resolveTranslog(), logger);
+        RemoteFsTranslog.download(
+            repository,
+            shardId,
+            getThreadPool(),
+            shardPath().resolveTranslog(),
+            indexSettings.getRemoteStorePathType(),
+            logger
+        );
     }
 
     /**

server/src/main/java/org/opensearch/index/shard/StoreRecovery.java

@@ -58,6 +58,7 @@
 import org.opensearch.index.engine.Engine;
 import org.opensearch.index.engine.EngineException;
 import org.opensearch.index.mapper.MapperService;
+import org.opensearch.index.remote.RemoteStorePathType;
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.snapshots.IndexShardRestoreFailedException;
 import org.opensearch.index.snapshots.blobstore.RemoteStoreShardShallowCopySnapshot;
@@ -409,7 +410,8 @@ void recoverFromSnapshotAndRemoteStore(
                 RemoteSegmentStoreDirectory sourceRemoteDirectory = (RemoteSegmentStoreDirectory) directoryFactory.newDirectory(
                     remoteStoreRepository,
                     indexUUID,
-                    shardId
+                    shardId,
+                    RemoteStorePathType.FIXED // TODO - The path type needs to be obtained from RemoteStoreShardShallowCopySnapshot
                 );
                 sourceRemoteDirectory.initializeToSpecificCommit(
                     primaryTerm,

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java

@@ -31,6 +31,7 @@
 import org.opensearch.common.lucene.store.ByteArrayIndexInput;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.index.shard.ShardId;
+import org.opensearch.index.remote.RemoteStorePathType;
 import org.opensearch.index.remote.RemoteStoreUtils;
 import org.opensearch.index.store.lockmanager.FileLockInfo;
 import org.opensearch.index.store.lockmanager.RemoteStoreCommitLevelLockManager;
@@ -897,13 +898,15 @@ public static void remoteDirectoryCleanup(
         RemoteSegmentStoreDirectoryFactory remoteDirectoryFactory,
         String remoteStoreRepoForIndex,
         String indexUUID,
-        ShardId shardId
+        ShardId shardId,
+        RemoteStorePathType pathType
     ) {
         try {
             RemoteSegmentStoreDirectory remoteSegmentStoreDirectory = (RemoteSegmentStoreDirectory) remoteDirectoryFactory.newDirectory(
                 remoteStoreRepoForIndex,
                 indexUUID,
-                shardId
+                shardId,
+                pathType
             );
             remoteSegmentStoreDirectory.deleteStaleSegments(0);
             remoteSegmentStoreDirectory.deleteIfEmpty();