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

server/src/main/java/org/opensearch/index/translog/BufferedChecksumStreamInput.java → libs/core/src/main/java/org/opensearch/core/common/io/stream/BufferedChecksumStreamInput.java

@@ -30,12 +30,10 @@
  * GitHub history for details.
  */
 
-package org.opensearch.index.translog;
+package org.opensearch.core.common.io.stream;
 
 import org.apache.lucene.store.BufferedChecksum;
 import org.apache.lucene.util.BitUtil;
-import org.opensearch.core.common.io.stream.FilterStreamInput;
-import org.opensearch.core.common.io.stream.StreamInput;
 
 import java.io.EOFException;
 import java.io.IOException;

server/src/main/java/org/opensearch/index/translog/BufferedChecksumStreamOutput.java → libs/core/src/main/java/org/opensearch/core/common/io/stream/BufferedChecksumStreamOutput.java

@@ -30,11 +30,10 @@
  * GitHub history for details.
  */
 
-package org.opensearch.index.translog;
+package org.opensearch.core.common.io.stream;
 
 import org.apache.lucene.store.BufferedChecksum;
 import org.opensearch.common.annotation.PublicApi;
-import org.opensearch.core.common.io.stream.StreamOutput;
 
 import java.io.IOException;
 import java.util.zip.CRC32;

server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java

@@ -187,7 +187,8 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
         DiscoveryNode dn = remoteDN.orElseGet(() -> (currentNodes.getNodes().values()).stream().findFirst().get());
         RepositoriesMetadata repositoriesMetadata = remoteStoreNodeService.updateRepositoriesMetadata(
             dn,
-            currentState.getMetadata().custom(RepositoriesMetadata.TYPE)
+            currentState.getMetadata().custom(RepositoriesMetadata.TYPE),
+            currentState.getMetadata().settings()
         );
 
         assert nodesBuilder.isLocalNodeElectedClusterManager();
@@ -223,7 +224,8 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
                         logger.info("Updating system repository now for remote store");
                         repositoriesMetadata = remoteStoreNodeService.updateRepositoriesMetadata(
                             node,
-                            currentState.getMetadata().custom(RepositoriesMetadata.TYPE)
+                            currentState.getMetadata().custom(RepositoriesMetadata.TYPE),
+                            currentState.getMetadata().settings()
                         );
                     }
 
@@ -515,7 +517,7 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
 
             DiscoveryNode existingNode = existingNodes.get(0);
             if (joiningNode.isRemoteStoreNode()) {
-                ensureRemoteStoreNodesCompatibility(joiningNode, existingNode);
+                ensureRemoteStoreNodesCompatibility(joiningNode, existingNode, metadata.settings());
             } else {
                 if (existingNode.isRemoteStoreNode()) {
                     throw new IllegalStateException(
@@ -538,18 +540,20 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
                 }
                 if (joiningNode.isRemoteStoreNode()) {
                     Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
-                    remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));
+                    remoteDN.ifPresent(
+                        discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode, metadata.settings())
+                    );
                 }
             }
         }
     }
 
-    private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNode, DiscoveryNode existingNode) {
+    private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNode, DiscoveryNode existingNode, Settings settings) {
         if (joiningNode.isRemoteStoreNode()) {
             if (existingNode.isRemoteStoreNode()) {
-                RemoteStoreNodeAttribute joiningRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(joiningNode);
-                RemoteStoreNodeAttribute existingRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(existingNode);
-                if (existingRemoteStoreNodeAttribute.equals(joiningRemoteStoreNodeAttribute) == false) {
+                RemoteStoreNodeAttribute joiningRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(joiningNode, settings);
+                RemoteStoreNodeAttribute existingRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(existingNode, settings);
+                if (existingRemoteStoreNodeAttribute.equalsIgnoreOptionalRepo(joiningRemoteStoreNodeAttribute) == false) {
                     throw new IllegalStateException(
                         "a remote store node ["
                             + joiningNode

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

@@ -53,6 +53,7 @@
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.opensearch.repositories.blobstore.BlobStoreRepository.SYSTEM_REPOSITORY_SETTING;
 
@@ -164,6 +165,34 @@ public boolean equalsIgnoreGenerations(@Nullable RepositoriesMetadata other) {
         return true;
     }
 
+    /**
+     * Checks if this instance and the give instance share the same repositories, with option to skip checking for a list of repos.
+     * This will support
+     * @param other other repositories metadata
+     * @param reposToSkip list of repos to skip check for equality
+     * @return {@code true} iff both instances contain the same repositories apart from differences in generations, not including repos provided in reposToSkip.
+     */
+    public boolean equalsIgnoreGenerationsIgnoreOptionalRepos(@Nullable RepositoriesMetadata other, List<String> reposToSkip) {
+        if (other == null) {
+            return false;
+        }
+        List<RepositoryMetadata> currentRepositories = repositories.stream()
+            .filter(repo -> !reposToSkip.contains(repo.name()))
+            .collect(Collectors.toList());
+        List<RepositoryMetadata> otherRepositories = other.repositories.stream()
+            .filter(repo -> !reposToSkip.contains(repo.name()))
+            .collect(Collectors.toList());
+        if (otherRepositories.size() != currentRepositories.size()) {
+            return false;
+        }
+        for (int i = 0; i < currentRepositories.size(); i++) {
+            if (currentRepositories.get(i).equalsIgnoreGenerations(otherRepositories.get(i)) == false) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @Override
     public int hashCode() {
         return repositories.hashCode();

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

@@ -43,6 +43,7 @@
 import org.opensearch.common.util.set.Sets;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.util.CollectionUtils;
 import org.opensearch.core.index.Index;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.node.ResponseCollectorService;
@@ -738,7 +739,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

@@ -0,0 +1,237 @@
+/*
+ * 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.cluster.routing.remote;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.cluster.DiffableUtils;
+import org.opensearch.cluster.routing.IndexRoutingTable;
+import org.opensearch.cluster.routing.RoutingTable;
+import org.opensearch.common.blobstore.transfer.RemoteTransferContainer;
+import org.opensearch.common.io.stream.BytesStreamOutput;
+import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.io.IOUtils;
+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.RemoteClusterStateService;
+import org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable;
+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.function.Supplier;
+
+import org.apache.lucene.store.IndexInput;
+import org.opensearch.action.LatchedActionListener;
+import org.opensearch.cluster.ClusterState;
+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.stream.OffsetRangeIndexInputStream;
+
+import org.opensearch.common.lucene.store.ByteArrayIndexInput;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.index.remote.RemoteStoreEnums;
+import org.opensearch.index.remote.RemoteStorePathStrategy;
+import org.opensearch.index.remote.RemoteStoreUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
+
+/**
+ * A Service which provides APIs to upload and download routing table from remote store.
+ *
+ * @opensearch.internal
+ */
+public class RemoteRoutingTableService extends AbstractLifecycleComponent {
+
+    /**
+     * Cluster setting to specify if routing table should be published to remote store
+     */
+    public static final Setting<Boolean> REMOTE_ROUTING_TABLE_ENABLED_SETTING = Setting.boolSetting(
+        "cluster.remote_store.routing_table.enabled",
+        false,
+        Setting.Property.NodeScope,
+        Setting.Property.Final
+    );
+
+    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;
+    private BlobStoreRepository blobStoreRepository;
+
+    public RemoteRoutingTableService(Supplier<RepositoriesService> repositoriesService, Settings settings) {
+        assert isRemoteRoutingTableEnabled(settings) : "Remote routing table is not enabled";
+        this.repositoriesService = repositoriesService;
+        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> indicesRoutingToUpload, Set<String> indicesRoutingToDelete) {
+        final Map<String, ClusterMetadataManifest.UploadedIndexMetadata> allUploadedIndicesRouting = previousManifest.getIndicesRouting()
+            .stream()
+            .collect(Collectors.toMap(ClusterMetadataManifest.UploadedIndexMetadata::getIndexName, Function.identity()));
+
+        indicesRoutingToUpload.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) {
+                    throw new IOException("Failed to write IndexRoutingTable to remote store. ", 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) {
+                throw new IOException("Failed to write IndexRoutingTable to remote store. ", e);
+            }
+        } catch (IOException e) {
+            throw new IOException("Failed to create transfer object for IndexRoutingTable for remote store upload. ", 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) {
+            IOUtils.close(blobStoreRepository);
+        }
+    }
+
+    @Override
+    protected void doStart() {
+        assert isRemoteRoutingTableEnabled(settings) == true : "Remote routing table is not enabled";
+        final String remoteStoreRepo = settings.get(
+            Node.NODE_ATTRIBUTES.getKey() + RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY
+        );
+        assert remoteStoreRepo != null : "Remote routing table repository is not configured";
+        final Repository repository = repositoriesService.get().repository(remoteStoreRepo);
+        assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository";
+        blobStoreRepository = (BlobStoreRepository) repository;
+    }
+
+    @Override
+    protected void doStop() {}
+
+}

server/src/main/java/org/opensearch/cluster/routing/remote/package-info.java

@@ -0,0 +1,10 @@
+/*
+ * 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 containing class to perform operations on remote routing table */
+package org.opensearch.cluster.routing.remote;