Merge branch 'read_write_ephemeral' of github.com:shiv0408/OpenSearch…

… into read_write_ephemeral

server/src/main/java/org/opensearch/common/remote/AbstractRemoteWritableBlobEntity.java

@@ -42,6 +42,8 @@ public AbstractRemoteWritableBlobEntity(
 
     public abstract BlobPathParameters getBlobPathParameters();
 
+    public abstract String getType();
+
     public String getFullBlobName() {
         return blobName;
     }

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -528,7 +528,6 @@ private UploadedMetadataResults writeMetadataInParallel(
                         blobStoreRepository.getCompressor(),
                         blobStoreRepository.getNamedXContentRegistry()
                     ),
-                    remoteGlobalMetadataManager.getPersistentSettingsBlobStore(),
                     listener
                 )
             );
@@ -556,7 +555,6 @@ private UploadedMetadataResults writeMetadataInParallel(
                         blobStoreRepository.getCompressor(),
                         blobStoreRepository.getNamedXContentRegistry()
                     ),
-                    remoteGlobalMetadataManager.getCoordinationMetadataBlobStore(),
                     listener
                 )
             );
@@ -572,7 +570,6 @@ private UploadedMetadataResults writeMetadataInParallel(
                         blobStoreRepository.getCompressor(),
                         blobStoreRepository.getNamedXContentRegistry()
                     ),
-                    remoteGlobalMetadataManager.getTemplatesMetadataBlobStore(),
                     listener
                 )
             );
@@ -624,7 +621,6 @@ private UploadedMetadataResults writeMetadataInParallel(
                         blobStoreRepository.getCompressor(),
                         blobStoreRepository.getNamedXContentRegistry()
                     ),
-                    remoteGlobalMetadataManager.getCustomMetadataBlobStore(),
                     listener
                 )
             );

server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java

@@ -65,13 +65,7 @@ public class RemoteGlobalMetadataManager {
     public static final int GLOBAL_METADATA_CURRENT_CODEC_VERSION = 1;
 
     private volatile TimeValue globalMetadataUploadTimeout;
-    private final RemoteWritableEntityStore<Metadata, RemoteGlobalMetadata> globalMetadataBlobStore;
-    private final RemoteClusterStateBlobStore<CoordinationMetadata, RemoteCoordinationMetadata> coordinationMetadataBlobStore;
-    private final RemoteClusterStateBlobStore<Settings, RemoteTransientSettingsMetadata> transientSettingsBlobStore;
-    private final RemoteClusterStateBlobStore<Settings, RemotePersistentSettingsMetadata> persistentSettingsBlobStore;
-    private final RemoteClusterStateBlobStore<TemplatesMetadata, RemoteTemplatesMetadata> templatesMetadataBlobStore;
-    private final RemoteClusterStateBlobStore<Custom, RemoteCustomMetadata> customMetadataBlobStore;
-    private final RemoteClusterStateBlobStore<DiffableStringMap, RemoteHashesOfConsistentSettings> hashesOfConsistentSettingsBlobStore;
+    private Map<String, RemoteWritableEntityStore> remoteWritableEntityStores;
     private final Compressor compressor;
     private final NamedXContentRegistry namedXContentRegistry;
 
@@ -85,40 +79,56 @@ public class RemoteGlobalMetadataManager {
         this.globalMetadataUploadTimeout = clusterSettings.get(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING);
         this.compressor = blobStoreRepository.getCompressor();
         this.namedXContentRegistry = blobStoreRepository.getNamedXContentRegistry();
-        globalMetadataBlobStore = new RemoteClusterStateBlobStore<>(
-            blobStoreTransferService,
-            blobStoreRepository,
-            clusterName,
-            threadpool,
-            ThreadPool.Names.REMOTE_STATE_READ
+        this.remoteWritableEntityStores = new HashMap<>();
+        this.remoteWritableEntityStores.put(
+            RemoteGlobalMetadata.GLOBAL_METADATA,
+            new RemoteClusterStateBlobStore<>(
+                blobStoreTransferService,
+                blobStoreRepository,
+                clusterName,
+                threadpool,
+                ThreadPool.Names.REMOTE_STATE_READ
+            )
         );
-        coordinationMetadataBlobStore = new RemoteClusterStateBlobStore<>(
-            blobStoreTransferService,
-            blobStoreRepository,
-            clusterName,
-            threadpool,
-            ThreadPool.Names.REMOTE_STATE_READ
+        this.remoteWritableEntityStores.put(
+            RemoteCoordinationMetadata.COORDINATION_METADATA,
+            new RemoteClusterStateBlobStore<>(
+                blobStoreTransferService,
+                blobStoreRepository,
+                clusterName,
+                threadpool,
+                ThreadPool.Names.REMOTE_STATE_READ
+            )
         );
-        persistentSettingsBlobStore = new RemoteClusterStateBlobStore<>(
-            blobStoreTransferService,
-            blobStoreRepository,
-            clusterName,
-            threadpool,
-            ThreadPool.Names.REMOTE_STATE_READ
+        this.remoteWritableEntityStores.put(
+            RemotePersistentSettingsMetadata.SETTING_METADATA,
+            new RemoteClusterStateBlobStore<>(
+                blobStoreTransferService,
+                blobStoreRepository,
+                clusterName,
+                threadpool,
+                ThreadPool.Names.REMOTE_STATE_READ
+            )
         );
-        templatesMetadataBlobStore = new RemoteClusterStateBlobStore<>(
-            blobStoreTransferService,
-            blobStoreRepository,
-            clusterName,
-            threadpool,
-            ThreadPool.Names.REMOTE_STATE_READ
+        this.remoteWritableEntityStores.put(
+            RemoteTemplatesMetadata.TEMPLATES_METADATA,
+            new RemoteClusterStateBlobStore<>(
+                blobStoreTransferService,
+                blobStoreRepository,
+                clusterName,
+                threadpool,
+                ThreadPool.Names.REMOTE_STATE_READ
+            )
         );
-        customMetadataBlobStore = new RemoteClusterStateBlobStore<>(
-            blobStoreTransferService,
-            blobStoreRepository,
-            clusterName,
-            threadpool,
-            ThreadPool.Names.REMOTE_STATE_READ
+        this.remoteWritableEntityStores.put(
+            RemoteCustomMetadata.CUSTOM_METADATA,
+            new RemoteClusterStateBlobStore<>(
+                blobStoreTransferService,
+                blobStoreRepository,
+                clusterName,
+                threadpool,
+                ThreadPool.Names.REMOTE_STATE_READ
+            )
         );
         clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setGlobalMetadataUploadTimeout);
     }
@@ -128,10 +138,20 @@ public class RemoteGlobalMetadataManager {
      */
     CheckedRunnable<IOException> getAsyncMetadataWriteAction(
         AbstractRemoteWritableBlobEntity writeEntity,
-        RemoteWritableEntityStore remoteStore,
         LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener
     ) {
-        return () -> remoteStore.writeAsync(writeEntity, getActionListener(writeEntity, latchedActionListener));
+        return (() -> getRemoteWriteableEntityStoreForObject(writeEntity).writeAsync(
+            writeEntity,
+            getActionListener(writeEntity, latchedActionListener)
+        ));
+    }
+
+    private RemoteWritableEntityStore getRemoteWriteableEntityStoreForObject(AbstractRemoteWritableBlobEntity entity) {
+        RemoteWritableEntityStore remoteStore = remoteWritableEntityStores.get(entity.getType());
+        if (remoteStore == null) {
+            throw new IllegalArgumentException("Unknown entity type [" + entity.getType() + "]");
+        }
+        return remoteStore;
     }
 
     private ActionListener<Void> getActionListener(
@@ -188,7 +208,7 @@ Metadata getGlobalMetadata(String clusterUUID, ClusterMetadataManifest clusterMe
                     compressor,
                     namedXContentRegistry
                 );
-                return globalMetadataBlobStore.read(remoteGlobalMetadata);
+                return (Metadata) getRemoteWriteableEntityStoreForObject(remoteGlobalMetadata).read(remoteGlobalMetadata);
             } else if (clusterMetadataManifest.hasMetadataAttributesFiles()) {
                 Metadata.Builder builder = new Metadata.Builder();
                 if (clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename() != null) {
@@ -198,7 +218,11 @@ Metadata getGlobalMetadata(String clusterUUID, ClusterMetadataManifest clusterMe
                         compressor,
                         namedXContentRegistry
                     );
-                    builder.coordinationMetadata(coordinationMetadataBlobStore.read(remoteCoordinationMetadata));
+                    builder.coordinationMetadata(
+                        (CoordinationMetadata) getRemoteWriteableEntityStoreForObject(remoteCoordinationMetadata).read(
+                            remoteCoordinationMetadata
+                        )
+                    );
                 }
                 if (clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename() != null) {
                     RemoteTemplatesMetadata remoteTemplatesMetadata = new RemoteTemplatesMetadata(
@@ -207,7 +231,9 @@ Metadata getGlobalMetadata(String clusterUUID, ClusterMetadataManifest clusterMe
                         compressor,
                         namedXContentRegistry
                     );
-                    builder.templates(templatesMetadataBlobStore.read(remoteTemplatesMetadata));
+                    builder.templates(
+                        (TemplatesMetadata) getRemoteWriteableEntityStoreForObject(remoteTemplatesMetadata).read(remoteTemplatesMetadata)
+                    );
                 }
                 if (clusterMetadataManifest.getSettingsMetadata().getUploadedFilename() != null) {
                     RemotePersistentSettingsMetadata remotePersistentSettingsMetadata = new RemotePersistentSettingsMetadata(
@@ -216,18 +242,27 @@ Metadata getGlobalMetadata(String clusterUUID, ClusterMetadataManifest clusterMe
                         compressor,
                         namedXContentRegistry
                     );
-                    builder.persistentSettings(persistentSettingsBlobStore.read(remotePersistentSettingsMetadata));
+                    builder.persistentSettings(
+                        (Settings) getRemoteWriteableEntityStoreForObject(remotePersistentSettingsMetadata).read(
+                            remotePersistentSettingsMetadata
+                        )
+                    );
                 }
                 builder.clusterUUID(clusterMetadataManifest.getClusterUUID());
                 builder.clusterUUIDCommitted(clusterMetadataManifest.isClusterUUIDCommitted());
                 builder.version(clusterMetadataManifest.getMetadataVersion());
                 clusterMetadataManifest.getCustomMetadataMap().forEach((key, value) -> {
                     try {
+                        RemoteCustomMetadata remoteCustomMetadata = new RemoteCustomMetadata(
+                            value.getUploadedFilename(),
+                            key,
+                            clusterUUID,
+                            compressor,
+                            namedXContentRegistry
+                        );
                         builder.putCustom(
                             key,
-                            customMetadataBlobStore.read(
-                                new RemoteCustomMetadata(value.getUploadedFilename(), key, clusterUUID, compressor, namedXContentRegistry)
-                            )
+                            (Custom) getRemoteWriteableEntityStoreForObject(remoteCustomMetadata).read(remoteCustomMetadata)
                         );
                     } catch (IOException e) {
                         throw new IllegalStateException(
@@ -286,20 +321,4 @@ private void setGlobalMetadataUploadTimeout(TimeValue newGlobalMetadataUploadTim
     public TimeValue getGlobalMetadataUploadTimeout() {
         return this.globalMetadataUploadTimeout;
     }
-
-    public RemoteClusterStateBlobStore<CoordinationMetadata, RemoteCoordinationMetadata> getCoordinationMetadataBlobStore() {
-        return coordinationMetadataBlobStore;
-    }
-
-    public RemoteClusterStateBlobStore<Settings, RemotePersistentSettingsMetadata> getPersistentSettingsBlobStore() {
-        return persistentSettingsBlobStore;
-    }
-
-    public RemoteClusterStateBlobStore<TemplatesMetadata, RemoteTemplatesMetadata> getTemplatesMetadataBlobStore() {
-        return templatesMetadataBlobStore;
-    }
-
-    public RemoteClusterStateBlobStore<Custom, RemoteCustomMetadata> getCustomMetadataBlobStore() {
-        return customMetadataBlobStore;
-    }
 }

server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterBlocks.java

@@ -55,6 +55,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN), CLUSTER_BLOCKS);
     }
 
+    @Override
+    public String getType() {
+        return CLUSTER_BLOCKS;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/transient/<componentPrefix>__<inverted_state_version>__<inverted__timestamp>__<codec_version>

server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterMetadataManifest.java

@@ -86,6 +86,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(MANIFEST), MANIFEST);
     }
 
+    @Override
+    public String getType() {
+        return MANIFEST;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest__<inverted_term>__<inverted_version>__C/P__<inverted__timestamp>__
@@ -150,4 +155,5 @@ private ChecksumBlobStoreFormat<ClusterMetadataManifest> getClusterMetadataManif
         }
         throw new IllegalArgumentException("Cluster metadata manifest file is corrupted, don't have valid codec version");
     }
+
 }

server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterStateCustoms.java

@@ -77,6 +77,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN), prefix);
     }
 
+    @Override
+    public String getType() {
+        return CLUSTER_STATE_CUSTOM;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/ephemeral/<componentPrefix>__<inverted_state_version>__<inverted__timestamp>__<codec_version>

server/src/main/java/org/opensearch/gateway/remote/model/RemoteCoordinationMetadata.java

@@ -70,6 +70,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of("global-metadata"), COORDINATION_METADATA);
     }
 
+    @Override
+    public String getType() {
+        return COORDINATION_METADATA;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/<componentPrefix>__<inverted_metadata_version>__<inverted__timestamp>__<codec_version>

server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java

@@ -85,6 +85,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(GLOBAL_METADATA_PATH_TOKEN), prefix);
     }
 
+    @Override
+    public String getType() {
+        return CUSTOM_METADATA;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/<componentPrefix>__<inverted_metadata_version>__<inverted__timestamp>__

server/src/main/java/org/opensearch/gateway/remote/model/RemoteDiscoveryNodes.java

@@ -60,6 +60,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN), DISCOVERY_NODES);
     }
 
+    @Override
+    public String getType() {
+        return DISCOVERY_NODES;
+    }
+
     @Override
     public String generateBlobFileName() {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/ephemeral/<componentPrefix>__<inverted_state_version>__<inverted__timestamp>__<codec_version>

server/src/main/java/org/opensearch/gateway/remote/model/RemoteGlobalMetadata.java

@@ -26,6 +26,7 @@
  * Wrapper class for uploading/downloading global metadata ({@link Metadata}) to/from remote blob store
  */
 public class RemoteGlobalMetadata extends AbstractRemoteWritableBlobEntity<Metadata> {
+    public static final String GLOBAL_METADATA = "global_metadata";
 
     public static final ChecksumBlobStoreFormat<Metadata> GLOBAL_METADATA_FORMAT = new ChecksumBlobStoreFormat<>(
         "metadata",
@@ -48,6 +49,11 @@ public BlobPathParameters getBlobPathParameters() {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public String getType() {
+        return GLOBAL_METADATA;
+    }
+
     @Override
     public String generateBlobFileName() {
         throw new UnsupportedOperationException();

server/src/main/java/org/opensearch/gateway/remote/model/RemoteHashesOfConsistentSettings.java

@@ -58,6 +58,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of(GLOBAL_METADATA_PATH_TOKEN), HASHES_OF_CONSISTENT_SETTINGS);
     }
 
+    @Override
+    public String getType() {
+        return HASHES_OF_CONSISTENT_SETTINGS;
+    }
+
     @Override
     public String generateBlobFileName() {
         String blobFileName = String.join(

server/src/main/java/org/opensearch/gateway/remote/model/RemoteIndexMetadata.java

@@ -38,7 +38,7 @@ public class RemoteIndexMetadata extends AbstractRemoteWritableBlobEntity<IndexM
         METADATA_NAME_PLAIN_FORMAT,
         IndexMetadata::fromXContent
     );
-    public static final String INDEX_PATH_TOKEN = "index";
+    public static final String INDEX = "index";
 
     private IndexMetadata indexMetadata;
 
@@ -64,7 +64,12 @@ public RemoteIndexMetadata(
 
     @Override
     public BlobPathParameters getBlobPathParameters() {
-        return new BlobPathParameters(List.of(INDEX_PATH_TOKEN, indexMetadata.getIndexUUID()), "metadata");
+        return new BlobPathParameters(List.of(INDEX, indexMetadata.getIndexUUID()), "metadata");
+    }
+
+    @Override
+    public String getType() {
+        return INDEX;
     }
 
     @Override

server/src/main/java/org/opensearch/gateway/remote/model/RemotePersistentSettingsMetadata.java

@@ -71,6 +71,11 @@ public BlobPathParameters getBlobPathParameters() {
         return new BlobPathParameters(List.of("global-metadata"), SETTING_METADATA);
     }
 
+    @Override
+    public String getType() {
+        return SETTING_METADATA;
+    }
+
     @Override
     public String generateBlobFileName() {
         String blobFileName = String.join(