[Remote Routing Table] Initial commit for RemoteRoutingTableService setup (#13304)

* Initial commit for RemoteRoutingTableService setup

Signed-off-by: Himshikha Gupta <himshikh@amazon.com>
(cherry picked from commit 5121409)

Author: himshikha

CHANGELOG.md

@@ -16,6 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Query Insights] Add exporter support for top n queries ([#12982](https://github.com/opensearch-project/OpenSearch/pull/12982))
 - [Query Insights] Add X-Opaque-Id to search request metadata for top n queries ([#13374](https://github.com/opensearch-project/OpenSearch/pull/13374))
 - [Remote State] Add async remote state deletion task running on an interval, configurable by a setting ([#13995](https://github.com/opensearch-project/OpenSearch/pull/13995))
+- Add remote routing table for remote state publication with experimental feature flag ([#13304](https://github.com/opensearch-project/OpenSearch/pull/13304))
 
 ### Dependencies
 - Bump `com.github.spullara.mustache.java:compiler` from 0.9.10 to 0.9.13 ([#13329](https://github.com/opensearch-project/OpenSearch/pull/13329), [#13559](https://github.com/opensearch-project/OpenSearch/pull/13559))

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

@@ -572,11 +572,27 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
         assert existingNodes.isEmpty() == false;
 
         CompatibilityMode remoteStoreCompatibilityMode = REMOTE_STORE_COMPATIBILITY_MODE_SETTING.get(metadata.settings());
-        if (STRICT.equals(remoteStoreCompatibilityMode)) {
 
-            DiscoveryNode existingNode = existingNodes.get(0);
+        List<String> reposToSkip = new ArrayList<>(1);
+        Optional<DiscoveryNode> remoteRoutingTableNode = existingNodes.stream()
+            .filter(
+                node -> node.getAttributes().get(RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY) != null
+            )
+            .findFirst();
+        // If none of the existing nodes have routing table repo, then we skip this repo check if present in joining node.
+        // This ensures a new node with remote routing table repo is able to join the cluster.
+        if (remoteRoutingTableNode.isEmpty()) {
+            String joiningNodeRepoName = joiningNode.getAttributes()
+                .get(RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY);
+            if (joiningNodeRepoName != null) {
+                reposToSkip.add(joiningNodeRepoName);
+            }
+        }
+
+        if (STRICT.equals(remoteStoreCompatibilityMode)) {
+            DiscoveryNode existingNode = remoteRoutingTableNode.orElseGet(() -> existingNodes.get(0));
             if (joiningNode.isRemoteStoreNode()) {
-                ensureRemoteStoreNodesCompatibility(joiningNode, existingNode);
+                ensureRemoteStoreNodesCompatibility(joiningNode, existingNode, reposToSkip);
             } else {
                 if (existingNode.isRemoteStoreNode()) {
                     throw new IllegalStateException(
@@ -598,19 +614,25 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
                     throw new IllegalStateException(reason);
                 }
                 if (joiningNode.isRemoteStoreNode()) {
-                    Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
-                    remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));
+                    Optional<DiscoveryNode> remoteDN = remoteRoutingTableNode.isPresent()
+                        ? remoteRoutingTableNode
+                        : existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
+                    remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode, reposToSkip));
                 }
             }
         }
     }
 
-    private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNode, DiscoveryNode existingNode) {
+    private static void ensureRemoteStoreNodesCompatibility(
+        DiscoveryNode joiningNode,
+        DiscoveryNode existingNode,
+        List<String> reposToSkip
+    ) {
         if (joiningNode.isRemoteStoreNode()) {
             if (existingNode.isRemoteStoreNode()) {
                 RemoteStoreNodeAttribute joiningRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(joiningNode);
                 RemoteStoreNodeAttribute existingRemoteStoreNodeAttribute = new RemoteStoreNodeAttribute(existingNode);
-                if (existingRemoteStoreNodeAttribute.equals(joiningRemoteStoreNodeAttribute) == false) {
+                if (existingRemoteStoreNodeAttribute.equalsWithRepoSkip(joiningRemoteStoreNodeAttribute, reposToSkip) == false) {
                     throw new IllegalStateException(
                         "a remote store node ["
                             + joiningNode

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

@@ -51,8 +51,10 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 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 +166,40 @@ 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 equalsIgnoreGenerationsWithRepoSkip(@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;
+        }
+        // Sort repos by name for ordered comparison
+        Comparator<RepositoryMetadata> compareByName = (o1, o2) -> o1.name().compareTo(o2.name());
+        currentRepositories.sort(compareByName);
+        otherRepositories.sort(compareByName);
+
+        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/remote/RemoteRoutingTableService.java

@@ -0,0 +1,67 @@
+/*
+ * 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.common.lifecycle.AbstractLifecycleComponent;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.io.IOUtils;
+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 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 {
+
+    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;
+    }
+
+    @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;

server/src/main/java/org/opensearch/common/settings/FeatureFlagSettings.java

@@ -36,6 +36,7 @@ protected FeatureFlagSettings(
         FeatureFlags.DATETIME_FORMATTER_CACHING_SETTING,
         FeatureFlags.TIERED_REMOTE_INDEX_SETTING,
         FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING,
-        FeatureFlags.PLUGGABLE_CACHE_SETTING
+        FeatureFlags.PLUGGABLE_CACHE_SETTING,
+        FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL_SETTING
     );
 }

server/src/main/java/org/opensearch/common/util/FeatureFlags.java

@@ -67,6 +67,11 @@ public class FeatureFlags {
      */
     public static final String PLUGGABLE_CACHE = "opensearch.experimental.feature.pluggable.caching.enabled";
 
+    /**
+     * Gates the functionality of remote routing table.
+     */
+    public static final String REMOTE_PUBLICATION_EXPERIMENTAL = "opensearch.experimental.feature.remote_store.publication.enabled";
+
     public static final Setting<Boolean> REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING = Setting.boolSetting(
         REMOTE_STORE_MIGRATION_EXPERIMENTAL,
         false,
@@ -89,14 +94,21 @@ public class FeatureFlags {
 
     public static final Setting<Boolean> PLUGGABLE_CACHE_SETTING = Setting.boolSetting(PLUGGABLE_CACHE, false, Property.NodeScope);
 
+    public static final Setting<Boolean> REMOTE_PUBLICATION_EXPERIMENTAL_SETTING = Setting.boolSetting(
+        REMOTE_PUBLICATION_EXPERIMENTAL,
+        false,
+        Property.NodeScope
+    );
+
     private static final List<Setting<Boolean>> ALL_FEATURE_FLAG_SETTINGS = List.of(
         REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING,
         EXTENSIONS_SETTING,
         IDENTITY_SETTING,
         TELEMETRY_SETTING,
         DATETIME_FORMATTER_CACHING_SETTING,
         TIERED_REMOTE_INDEX_SETTING,
-        PLUGGABLE_CACHE_SETTING
+        PLUGGABLE_CACHE_SETTING,
+        REMOTE_PUBLICATION_EXPERIMENTAL_SETTING
     );
     /**
      * Should store the settings from opensearch.yml.

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

@@ -19,6 +19,7 @@
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.metadata.TemplatesMetadata;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.cluster.routing.remote.RemoteRoutingTableService;
 import org.opensearch.common.CheckedRunnable;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.blobstore.BlobContainer;
@@ -69,6 +70,7 @@
 
 import static java.util.Objects.requireNonNull;
 import static org.opensearch.gateway.PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled;
 
 /**
@@ -201,6 +203,7 @@ public class RemoteClusterStateService implements Closeable {
     private final List<IndexMetadataUploadListener> indexMetadataUploadListeners;
     private BlobStoreRepository blobStoreRepository;
     private BlobStoreTransferService blobStoreTransferService;
+    private Optional<RemoteRoutingTableService> remoteRoutingTableService;
     private volatile TimeValue slowWriteLoggingThreshold;
 
     private volatile TimeValue indexMetadataUploadTimeout;
@@ -253,6 +256,9 @@ public RemoteClusterStateService(
         this.remoteStateStats = new RemotePersistenceStats();
         this.remoteClusterStateCleanupManager = new RemoteClusterStateCleanupManager(this, clusterService);
         this.indexMetadataUploadListeners = indexMetadataUploadListeners;
+        this.remoteRoutingTableService = isRemoteRoutingTableEnabled(settings)
+            ? Optional.of(new RemoteRoutingTableService(repositoriesService, settings))
+            : Optional.empty();
     }
 
     /**
@@ -749,6 +755,9 @@ public void close() throws IOException {
         if (blobStoreRepository != null) {
             IOUtils.close(blobStoreRepository);
         }
+        if (this.remoteRoutingTableService.isPresent()) {
+            this.remoteRoutingTableService.get().close();
+        }
     }
 
     public void start() {
@@ -761,6 +770,7 @@ public void start() {
         assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository";
         blobStoreRepository = (BlobStoreRepository) repository;
         remoteClusterStateCleanupManager.start();
+        this.remoteRoutingTableService.ifPresent(RemoteRoutingTableService::start);
     }
 
     private ClusterMetadataManifest uploadManifest(
@@ -945,6 +955,11 @@ public TimeValue getMetadataManifestUploadTimeout() {
         return this.metadataManifestUploadTimeout;
     }
 
+    // Package private for unit test
+    Optional<RemoteRoutingTableService> getRemoteRoutingTableService() {
+        return this.remoteRoutingTableService;
+    }
+
     static String getManifestFileName(long term, long version, boolean committed, int codecVersion) {
         // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest__<inverted_term>__<inverted_version>__C/P__<inverted__timestamp>__<codec_version>
         return String.join(

server/src/main/java/org/opensearch/node/remotestore/RemoteStoreNodeAttribute.java

@@ -13,6 +13,7 @@
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.gateway.remote.RemoteClusterStateService;
 import org.opensearch.node.Node;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
@@ -28,6 +29,8 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
+
 /**
  * This is an abstraction for validating and storing information specific to remote backed storage nodes.
  *
@@ -46,6 +49,8 @@ public class RemoteStoreNodeAttribute {
         + "."
         + CryptoMetadata.SETTINGS_KEY;
     public static final String REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX = "remote_store.repository.%s.settings.";
+    public static final String REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY = "remote_store.routing_table.repository";
+
     private final RepositoriesMetadata repositoriesMetadata;
 
     public static List<String> SUPPORTED_DATA_REPO_NAME_ATTRIBUTES = List.of(
@@ -157,6 +162,10 @@ private Set<String> getValidatedRepositoryNames(DiscoveryNode node) {
         } else if (node.getAttributes().containsKey(REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY)) {
             repositoryNames.add(validateAttributeNonNull(node, REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY));
         }
+        if (node.getAttributes().containsKey(REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY)) {
+            repositoryNames.add(validateAttributeNonNull(node, REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY));
+        }
+
         return repositoryNames;
     }
 
@@ -187,6 +196,15 @@ public static boolean isRemoteStoreClusterStateEnabled(Settings settings) {
             && isRemoteClusterStateAttributePresent(settings);
     }
 
+    private static boolean isRemoteRoutingTableAttributePresent(Settings settings) {
+        return settings.getByPrefix(Node.NODE_ATTRIBUTES.getKey() + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY)
+            .isEmpty() == false;
+    }
+
+    public static boolean isRemoteRoutingTableEnabled(Settings settings) {
+        return FeatureFlags.isEnabled(REMOTE_PUBLICATION_EXPERIMENTAL) && isRemoteRoutingTableAttributePresent(settings);
+    }
+
     public RepositoriesMetadata getRepositoriesMetadata() {
         return this.repositoriesMetadata;
     }
@@ -231,6 +249,21 @@ public int hashCode() {
         return hashCode;
     }
 
+    /**
+     * Checks if 2 instances are equal, with option to skip check for a list of repos.
+     * *
+     * @param o other instance
+     * @param reposToSkip list of repos to skip check for equality
+     * @return {@code true} iff both instances are equal, not including the repositories in both instances if they are part of reposToSkip.
+     */
+    public boolean equalsWithRepoSkip(Object o, List<String> reposToSkip) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        RemoteStoreNodeAttribute that = (RemoteStoreNodeAttribute) o;
+        return this.getRepositoriesMetadata().equalsIgnoreGenerationsWithRepoSkip(that.getRepositoriesMetadata(), reposToSkip);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;