[Remote Routing Table] Add write flow for remote routing table

server/src/main/java/org/opensearch/cluster/routing/IndexShardRoutingTable.java

@@ -738,7 +738,7 @@ public boolean equals(Object o) {
         IndexShardRoutingTable that = (IndexShardRoutingTable) o;
 
         if (!shardId.equals(that.shardId)) return false;
-        if (!shards.equals(that.shards)) return false;
+        if (!new HashSet<>(shards).equals(new HashSet<>(that.shards))) return false;
 
         return true;
     }

server/src/main/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableService.java

@@ -10,17 +10,47 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.store.IndexInput;
+import org.opensearch.action.LatchedActionListener;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.DiffableUtils;
+import org.opensearch.cluster.routing.IndexRoutingTable;
+import org.opensearch.cluster.routing.RoutingTable;
+import org.opensearch.common.CheckedRunnable;
+import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer;
+import org.opensearch.common.blobstore.BlobContainer;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.common.blobstore.stream.write.WritePriority;
+import org.opensearch.common.blobstore.transfer.RemoteTransferContainer;
+import org.opensearch.common.blobstore.transfer.stream.OffsetRangeIndexInputStream;
+import org.opensearch.common.io.stream.BytesStreamOutput;
 import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
+import org.opensearch.common.lucene.store.ByteArrayIndexInput;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.gateway.remote.RemoteClusterStateService;
+import org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable;
+import org.opensearch.index.remote.RemoteStoreEnums;
+import org.opensearch.index.remote.RemoteStorePathStrategy;
+import org.opensearch.index.remote.RemoteStoreUtils;
 import org.opensearch.node.Node;
 import org.opensearch.node.remotestore.RemoteStoreNodeAttribute;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 
@@ -31,6 +61,11 @@
  */
 public class RemoteRoutingTableService extends AbstractLifecycleComponent {
 
+    public static final String INDEX_ROUTING_PATH_TOKEN = "index-routing";
+    public static final String INDEX_ROUTING_FILE_PREFIX = "index_routing";
+    public static final String DELIMITER = "__";
+    public static final String INDEX_ROUTING_METADATA_PREFIX = "indexRouting--";
+
     private static final Logger logger = LogManager.getLogger(RemoteRoutingTableService.class);
     private final Settings settings;
     private final Supplier<RepositoriesService> repositoriesService;
@@ -42,6 +77,142 @@ public RemoteRoutingTableService(Supplier<RepositoriesService> repositoriesServi
         this.settings = settings;
     }
 
+    private static final DiffableUtils.NonDiffableValueSerializer<String, IndexRoutingTable> CUSTOM_ROUTING_TABLE_VALUE_SERIALIZER =
+        new DiffableUtils.NonDiffableValueSerializer<String, IndexRoutingTable>() {
+            @Override
+            public void write(IndexRoutingTable value, StreamOutput out) throws IOException {
+                value.writeTo(out);
+            }
+
+            @Override
+            public IndexRoutingTable read(StreamInput in, String key) throws IOException {
+                return IndexRoutingTable.readFrom(in);
+            }
+        };
+
+    public static DiffableUtils.MapDiff<String, IndexRoutingTable, Map<String, IndexRoutingTable>> getIndicesRoutingMapDiff(
+        RoutingTable before,
+        RoutingTable after
+    ) {
+        return DiffableUtils.diff(
+            before.getIndicesRouting(),
+            after.getIndicesRouting(),
+            DiffableUtils.getStringKeySerializer(),
+            CUSTOM_ROUTING_TABLE_VALUE_SERIALIZER
+        );
+    }
+
+    public CheckedRunnable<IOException> getIndexRoutingAsyncAction(
+        ClusterState clusterState,
+        IndexRoutingTable indexRouting,
+        LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener,
+        BlobPath clusterBasePath
+    ) throws IOException {
+
+        BlobPath indexRoutingPath = clusterBasePath.add(INDEX_ROUTING_PATH_TOKEN);
+        BlobPath path = RemoteStoreEnums.PathType.HASHED_PREFIX.path(
+            RemoteStorePathStrategy.BasePathInput.builder().basePath(indexRoutingPath).indexUUID(indexRouting.getIndex().getUUID()).build(),
+            RemoteStoreEnums.PathHashAlgorithm.FNV_1A_BASE64
+        );
+        final BlobContainer blobContainer = blobStoreRepository.blobStore().blobContainer(path);
+
+        final String fileName = getIndexRoutingFileName();
+
+        ActionListener<Void> completionListener = ActionListener.wrap(
+            resp -> latchedActionListener.onResponse(
+                new ClusterMetadataManifest.UploadedIndexMetadata(
+                    indexRouting.getIndex().getName(),
+                    indexRouting.getIndex().getUUID(),
+                    path.buildAsString() + fileName,
+                    INDEX_ROUTING_METADATA_PREFIX
+                )
+            ),
+            ex -> latchedActionListener.onFailure(
+                new RemoteClusterStateService.RemoteStateTransferException(
+                    "Exception in writing index to remote store: " + indexRouting.getIndex().toString(),
+                    ex
+                )
+            )
+        );
+
+        return () -> uploadIndex(indexRouting, fileName, blobContainer, completionListener);
+    }
+
+    public List<ClusterMetadataManifest.UploadedIndexMetadata> getAllUploadedIndicesRouting(
+        ClusterMetadataManifest previousManifest,
+        List<ClusterMetadataManifest.UploadedIndexMetadata> indicesRoutingUploaded,
+        List<String> indicesRoutingToDelete
+    ) {
+        final Map<String, ClusterMetadataManifest.UploadedIndexMetadata> allUploadedIndicesRouting = previousManifest.getIndicesRouting()
+            .stream()
+            .collect(Collectors.toMap(ClusterMetadataManifest.UploadedIndexMetadata::getIndexName, Function.identity()));
+
+        indicesRoutingUploaded.forEach(
+            uploadedIndexRouting -> allUploadedIndicesRouting.put(uploadedIndexRouting.getIndexName(), uploadedIndexRouting)
+        );
+        indicesRoutingToDelete.forEach(allUploadedIndicesRouting::remove);
+
+        return new ArrayList<>(allUploadedIndicesRouting.values());
+    }
+
+    private void uploadIndex(
+        IndexRoutingTable indexRouting,
+        String fileName,
+        BlobContainer blobContainer,
+        ActionListener<Void> completionListener
+    ) throws IOException {
+        RemoteIndexRoutingTable indexRoutingInput = new RemoteIndexRoutingTable(indexRouting);
+        BytesReference bytesInput = null;
+        try (BytesStreamOutput streamOutput = new BytesStreamOutput()) {
+            indexRoutingInput.writeTo(streamOutput);
+            bytesInput = streamOutput.bytes();
+        } catch (IOException e) {
+            throw new IOException("Failed to serialize IndexRoutingTable. ", e);
+        }
+
+        if (blobContainer instanceof AsyncMultiStreamBlobContainer == false) {
+            try {
+                blobContainer.writeBlob(fileName, bytesInput.streamInput(), bytesInput.length(), true);
+                completionListener.onResponse(null);
+            } catch (IOException e) {
+                logger.error("Failed to write IndexRoutingTable to remote store. ", e);
+                completionListener.onFailure(e);
+            }
+            return;
+        }
+
+        try (IndexInput input = new ByteArrayIndexInput("indexrouting", BytesReference.toBytes(bytesInput))) {
+            try (
+                RemoteTransferContainer remoteTransferContainer = new RemoteTransferContainer(
+                    fileName,
+                    fileName,
+                    input.length(),
+                    true,
+                    WritePriority.URGENT,
+                    (size, position) -> new OffsetRangeIndexInputStream(input, size, position),
+                    null,
+                    false
+                )
+            ) {
+                ((AsyncMultiStreamBlobContainer) blobContainer).asyncBlobUpload(
+                    remoteTransferContainer.createWriteContext(),
+                    completionListener
+                );
+            } catch (IOException e) {
+                logger.error("Failed to write IndexRoutingTable to remote store. ", e);
+                completionListener.onFailure(e);
+            }
+        } catch (IOException e) {
+            logger.error("Failed to create transfer object for IndexRoutingTable for remote store upload. ", e);
+            completionListener.onFailure(e);
+        }
+    }
+
+    private String getIndexRoutingFileName() {
+        return String.join(DELIMITER, INDEX_ROUTING_FILE_PREFIX, RemoteStoreUtils.invertLong(System.currentTimeMillis()));
+
+    }
+
     @Override
     protected void doClose() throws IOException {
         if (blobStoreRepository != null) {

server/src/main/java/org/opensearch/common/blobstore/BlobPath.java

@@ -39,6 +39,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * The list of paths where a blob can reside.  The contents of the paths are dependent upon the implementation of {@link BlobContainer}.
@@ -110,6 +111,19 @@ public BlobPath parent() {
         }
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        BlobPath strings = (BlobPath) o;
+        return Objects.equals(paths, strings.paths);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(paths);
+    }
+
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();

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

@@ -807,6 +807,7 @@ public static class UploadedIndexMetadata implements UploadedMetadata, Writeable
         private static final ParseField INDEX_NAME_FIELD = new ParseField("index_name");
         private static final ParseField INDEX_UUID_FIELD = new ParseField("index_uuid");
         private static final ParseField UPLOADED_FILENAME_FIELD = new ParseField("uploaded_filename");
+        private static final ParseField COMPONENT_PREFIX_FIELD = new ParseField("component_prefix");
 
         private static String indexName(Object[] fields) {
             return (String) fields[0];
@@ -820,23 +821,34 @@ private static String uploadedFilename(Object[] fields) {
             return (String) fields[2];
         }
 
+        private static String componentPrefix(Object[] fields) {
+            return (String) fields[3];
+        }
+
         private static final ConstructingObjectParser<UploadedIndexMetadata, Void> PARSER = new ConstructingObjectParser<>(
             "uploaded_index_metadata",
-            fields -> new UploadedIndexMetadata(indexName(fields), indexUUID(fields), uploadedFilename(fields))
+            fields -> new UploadedIndexMetadata(indexName(fields), indexUUID(fields), uploadedFilename(fields), componentPrefix(fields))
         );
 
         static {
             PARSER.declareString(ConstructingObjectParser.constructorArg(), INDEX_NAME_FIELD);
             PARSER.declareString(ConstructingObjectParser.constructorArg(), INDEX_UUID_FIELD);
             PARSER.declareString(ConstructingObjectParser.constructorArg(), UPLOADED_FILENAME_FIELD);
+            PARSER.declareString(ConstructingObjectParser.constructorArg(), COMPONENT_PREFIX_FIELD);
         }
 
         static final String COMPONENT_PREFIX = "index--";
+        private final String componentPrefix;
         private final String indexName;
         private final String indexUUID;
         private final String uploadedFilename;
 
         public UploadedIndexMetadata(String indexName, String indexUUID, String uploadedFileName) {
+            this(indexName, indexUUID, uploadedFileName, COMPONENT_PREFIX);
+        }
+
+        public UploadedIndexMetadata(String indexName, String indexUUID, String uploadedFileName, String componentPrefix) {
+            this.componentPrefix = componentPrefix;
             this.indexName = indexName;
             this.indexUUID = indexUUID;
             this.uploadedFilename = uploadedFileName;
@@ -846,6 +858,7 @@ public UploadedIndexMetadata(StreamInput in) throws IOException {
             this.indexName = in.readString();
             this.indexUUID = in.readString();
             this.uploadedFilename = in.readString();
+            this.componentPrefix = in.readString();
         }
 
         public String getUploadedFilePath() {
@@ -854,7 +867,7 @@ public String getUploadedFilePath() {
 
         @Override
         public String getComponent() {
-            return COMPONENT_PREFIX + getIndexName();
+            return componentPrefix + getIndexName();
         }
 
         public String getUploadedFilename() {
@@ -870,18 +883,24 @@ public String getIndexUUID() {
             return indexUUID;
         }
 
+        public String getComponentPrefix() {
+            return componentPrefix;
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             return builder.field(INDEX_NAME_FIELD.getPreferredName(), getIndexName())
                 .field(INDEX_UUID_FIELD.getPreferredName(), getIndexUUID())
-                .field(UPLOADED_FILENAME_FIELD.getPreferredName(), getUploadedFilePath());
+                .field(UPLOADED_FILENAME_FIELD.getPreferredName(), getUploadedFilePath())
+                .field(COMPONENT_PREFIX_FIELD.getPreferredName(), getComponentPrefix());
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(indexName);
             out.writeString(indexUUID);
             out.writeString(uploadedFilename);
+            out.writeString(componentPrefix);
         }
 
         @Override
@@ -895,12 +914,13 @@ public boolean equals(Object o) {
             final UploadedIndexMetadata that = (UploadedIndexMetadata) o;
             return Objects.equals(indexName, that.indexName)
                 && Objects.equals(indexUUID, that.indexUUID)
-                && Objects.equals(uploadedFilename, that.uploadedFilename);
+                && Objects.equals(uploadedFilename, that.uploadedFilename)
+                && Objects.equals(componentPrefix, that.componentPrefix);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(indexName, indexUUID, uploadedFilename);
+            return Objects.hash(indexName, indexUUID, uploadedFilename, componentPrefix);
         }
 
         @Override