Merge branch '6.x' into ccr-6.x

* 6.x:
  Allow engine to recover from translog upto a seqno (#33032)
  TEST: Skip assertSeqNos for closed shards (#33130)
  TEST: resync operation on replica should acquire shard permit (#33103)
  Add proxy support to RemoteClusterConnection (#33062)
  Build: Line up IDE detection logic
  Security index expands to a single replica (#33131)
  Suppress more tests
  HLRC: request/response homogeneity and JavaDoc improvements (#33133)
  [Rollup] Move toAggCap() methods out of rollup config objects (#32583)
  Muted all these tests due to #33128
  Fix race condition in scheduler engine test

Author: dnhatn

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/multi_cluster/20_info.yml

@@ -1,6 +1,8 @@
 ---
 "Fetch remote cluster info for existing cluster":
-
+  - skip:
+      version: "all"
+      reason: https://github.com/elastic/elasticsearch/issues/33128
   - do:
       cluster.remote_info: {}
   - match: { my_remote_cluster.connected: true }
@@ -11,6 +13,9 @@
 
 ---
 "Add transient remote cluster based on the preset cluster and check remote info":
+  - skip:
+      version: "all"
+      reason: https://github.com/elastic/elasticsearch/issues/33128
   - do:
       cluster.get_settings:
         include_defaults: true
@@ -65,8 +70,11 @@
 ---
 "skip_unavailable is returned as part of _remote/info response":
   - skip:
-      version: " - 6.0.99"
-      reason: "skip_unavailable is only returned from 6.1.0 on"
+      version: "all"
+      reason: https://github.com/elastic/elasticsearch/issues/33128
+  #- skip:
+  #    version: " - 6.0.99"
+  #    reason: "skip_unavailable is only returned from 6.1.0 on"
 
   - do:
       cluster.get_settings:

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -274,6 +274,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                     ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING,
                     TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
                     RemoteClusterAware.REMOTE_CLUSTERS_SEEDS,
+                    RemoteClusterAware.REMOTE_CLUSTERS_PROXY,
                     RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
                     RemoteClusterService.REMOTE_CONNECTIONS_PER_CLUSTER,
                     RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING,

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -1009,6 +1009,10 @@ public static Setting<String> simpleString(String key, Property... properties) {
         return new Setting<>(key, s -> "", Function.identity(), properties);
     }
 
+    public static Setting<String> simpleString(String key, Function<String, String> parser, Property... properties) {
+        return new Setting<>(key, s -> "", parser, properties);
+    }
+
     public static Setting<String> simpleString(String key, Setting<String> fallback, Property... properties) {
         return new Setting<>(key, fallback, Function.identity(), properties);
     }

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -1648,10 +1648,12 @@ public interface Warmer {
     public abstract int fillSeqNoGaps(long primaryTerm) throws IOException;
 
     /**
-     * Performs recovery from the transaction log.
+     * Performs recovery from the transaction log up to {@code recoverUpToSeqNo} (inclusive).
      * This operation will close the engine if the recovery fails.
+     *
+     * @param recoverUpToSeqNo the upper bound, inclusive, of sequence number to be recovered
      */
-    public abstract Engine recoverFromTranslog() throws IOException;
+    public abstract Engine recoverFromTranslog(long recoverUpToSeqNo) throws IOException;
 
     /**
      * Do not replay translog operations, but make the engine be ready.

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -398,15 +398,15 @@ private void bootstrapAppendOnlyInfoFromWriter(IndexWriter writer) {
     }
 
     @Override
-    public InternalEngine recoverFromTranslog() throws IOException {
+    public InternalEngine recoverFromTranslog(long recoverUpToSeqNo) throws IOException {
         flushLock.lock();
         try (ReleasableLock lock = readLock.acquire()) {
             ensureOpen();
             if (pendingTranslogRecovery.get() == false) {
                 throw new IllegalStateException("Engine has already been recovered");
             }
             try {
-                recoverFromTranslogInternal();
+                recoverFromTranslogInternal(recoverUpToSeqNo);
             } catch (Exception e) {
                 try {
                     pendingTranslogRecovery.set(true); // just play safe and never allow commits on this see #ensureCanFlush
@@ -428,11 +428,12 @@ public void skipTranslogRecovery() {
         pendingTranslogRecovery.set(false); // we are good - now we can commit
     }
 
-    private void recoverFromTranslogInternal() throws IOException {
+    private void recoverFromTranslogInternal(long recoverUpToSeqNo) throws IOException {
         Translog.TranslogGeneration translogGeneration = translog.getGeneration();
         final int opsRecovered;
-        final long translogGen = Long.parseLong(lastCommittedSegmentInfos.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
-        try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(translogGen)) {
+        final long translogFileGen = Long.parseLong(lastCommittedSegmentInfos.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
+        try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(
+            new Translog.TranslogGeneration(translog.getTranslogUUID(), translogFileGen), recoverUpToSeqNo)) {
             opsRecovered = config().getTranslogRecoveryRunner().run(this, snapshot);
         } catch (Exception e) {
             throw new EngineException(shardId, "failed to recover from translog", e);

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

@@ -1343,7 +1343,7 @@ int runTranslogRecovery(Engine engine, Translog.Snapshot snapshot) throws IOExce
      **/
     public void openEngineAndRecoverFromTranslog() throws IOException {
         innerOpenEngineAndTranslog();
-        getEngine().recoverFromTranslog();
+        getEngine().recoverFromTranslog(Long.MAX_VALUE);
     }
 
     /**

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -579,21 +579,27 @@ public long getLastSyncedGlobalCheckpoint() {
      */
     public Snapshot newSnapshot() throws IOException {
         try (ReleasableLock ignored = readLock.acquire()) {
-            return newSnapshotFromGen(getMinFileGeneration());
+            return newSnapshotFromGen(new TranslogGeneration(translogUUID, getMinFileGeneration()), Long.MAX_VALUE);
         }
     }
 
-    public Snapshot newSnapshotFromGen(long minGeneration) throws IOException {
+    public Snapshot newSnapshotFromGen(TranslogGeneration fromGeneration, long upToSeqNo) throws IOException {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
-            if (minGeneration < getMinFileGeneration()) {
-                throw new IllegalArgumentException("requested snapshot generation [" + minGeneration + "] is not available. " +
+            final long fromFileGen = fromGeneration.translogFileGeneration;
+            if (fromFileGen < getMinFileGeneration()) {
+                throw new IllegalArgumentException("requested snapshot generation [" + fromFileGen + "] is not available. " +
                     "Min referenced generation is [" + getMinFileGeneration() + "]");
             }
             TranslogSnapshot[] snapshots = Stream.concat(readers.stream(), Stream.of(current))
-                .filter(reader -> reader.getGeneration() >= minGeneration)
+                .filter(reader -> reader.getGeneration() >= fromFileGen && reader.getCheckpoint().minSeqNo <= upToSeqNo)
                 .map(BaseTranslogReader::newSnapshot).toArray(TranslogSnapshot[]::new);
-            return newMultiSnapshot(snapshots);
+            final Snapshot snapshot = newMultiSnapshot(snapshots);
+            if (upToSeqNo == Long.MAX_VALUE) {
+                return snapshot;
+            } else {
+                return new SeqNoFilterSnapshot(snapshot, Long.MIN_VALUE, upToSeqNo);
+            }
         }
     }
 
@@ -928,7 +934,59 @@ default int overriddenOperations() {
          * Returns the next operation in the snapshot or <code>null</code> if we reached the end.
          */
         Translog.Operation next() throws IOException;
+    }
+
+    /**
+     * A filtered snapshot consisting of only operations whose sequence numbers are in the given range
+     * between {@code fromSeqNo} (inclusive) and {@code toSeqNo} (inclusive). This filtered snapshot
+     * shares the same underlying resources with the {@code delegate} snapshot, therefore we should not
+     * use the {@code delegate} after passing it to this filtered snapshot.
+     */
+    static final class SeqNoFilterSnapshot implements Snapshot {
+        private final Snapshot delegate;
+        private int filteredOpsCount;
+        private final long fromSeqNo; // inclusive
+        private final long toSeqNo;   // inclusive
+
+        SeqNoFilterSnapshot(Snapshot delegate, long fromSeqNo, long toSeqNo) {
+            assert fromSeqNo <= toSeqNo : "from_seq_no[" + fromSeqNo + "] > to_seq_no[" + toSeqNo + "]";
+            this.delegate = delegate;
+            this.fromSeqNo = fromSeqNo;
+            this.toSeqNo = toSeqNo;
+        }
+
+        @Override
+        public int totalOperations() {
+            return delegate.totalOperations();
+        }
+
+        @Override
+        public int skippedOperations() {
+            return filteredOpsCount + delegate.skippedOperations();
+        }
 
+        @Override
+        public int overriddenOperations() {
+            return delegate.overriddenOperations();
+        }
+
+        @Override
+        public Operation next() throws IOException {
+            Translog.Operation op;
+            while ((op = delegate.next()) != null) {
+                if (fromSeqNo <= op.seqNo() && op.seqNo() <= toSeqNo) {
+                    return op;
+                } else {
+                    filteredOpsCount++;
+                }
+            }
+            return null;
+        }
+
+        @Override
+        public void close() throws IOException {
+            delegate.close();
+        }
     }
 
     /**

server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java

@@ -18,10 +18,14 @@
  */
 package org.elasticsearch.transport;
 
+import java.util.EnumSet;
 import java.util.function.Supplier;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.metadata.ClusterNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
@@ -66,6 +70,22 @@ public abstract class RemoteClusterAware extends AbstractComponent {
     public static final char REMOTE_CLUSTER_INDEX_SEPARATOR = ':';
     public static final String LOCAL_CLUSTER_GROUP_KEY = "";
 
+    /**
+     * A proxy address for the remote cluster.
+     * NOTE: this settings is undocumented until we have at last one transport that supports passing
+     * on the hostname via a mechanism like SNI.
+     */
+    public static final Setting.AffixSetting<String> REMOTE_CLUSTERS_PROXY = Setting.affixKeySetting(
+        "search.remote.",
+        "proxy",
+        key -> Setting.simpleString(key, s -> {
+            if (Strings.hasLength(s)) {
+                parsePort(s);
+            }
+            return s;
+        }, Setting.Property.NodeScope, Setting.Property.Dynamic), REMOTE_CLUSTERS_SEEDS);
+
+
     protected final ClusterNameExpressionResolver clusterNameResolver;
 
     /**
@@ -77,25 +97,42 @@ protected RemoteClusterAware(Settings settings) {
         this.clusterNameResolver = new ClusterNameExpressionResolver(settings);
     }
 
-    protected static Map<String, List<Supplier<DiscoveryNode>>> buildRemoteClustersSeeds(Settings settings) {
+    /**
+     * Builds the dynamic per-cluster config from the given settings. This is a map keyed by the cluster alias that points to a tuple
+     * (ProxyAddresss, [SeedNodeSuppliers]). If a cluster is configured with a proxy address all seed nodes will point to
+     * {@link TransportAddress#META_ADDRESS} and their configured address will be used as the hostname for the generated discovery node.
+     */
+    protected static Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> buildRemoteClustersDynamicConfig(Settings settings) {
         Stream<Setting<List<String>>> allConcreteSettings = REMOTE_CLUSTERS_SEEDS.getAllConcreteSettings(settings);
         return allConcreteSettings.collect(
             Collectors.toMap(REMOTE_CLUSTERS_SEEDS::getNamespace, concreteSetting -> {
                 String clusterName = REMOTE_CLUSTERS_SEEDS.getNamespace(concreteSetting);
                 List<String> addresses = concreteSetting.get(settings);
+                final boolean proxyMode = REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(clusterName).exists(settings);
                 List<Supplier<DiscoveryNode>> nodes = new ArrayList<>(addresses.size());
                 for (String address : addresses) {
-                    nodes.add(() -> {
-                        TransportAddress transportAddress = new TransportAddress(RemoteClusterAware.parseSeedAddress(address));
-                        return new DiscoveryNode(clusterName + "#" + transportAddress.toString(),
-                            transportAddress,
-                            Version.CURRENT.minimumCompatibilityVersion());
-                    });
+                    nodes.add(() -> buildSeedNode(clusterName, address, proxyMode));
                 }
-                return nodes;
+                return new Tuple<>(REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(clusterName).get(settings), nodes);
             }));
     }
 
+    static DiscoveryNode buildSeedNode(String clusterName, String address, boolean proxyMode) {
+        if (proxyMode) {
+            TransportAddress transportAddress = new TransportAddress(TransportAddress.META_ADDRESS, 0);
+            String hostName = address.substring(0, indexOfPortSeparator(address));
+            return new DiscoveryNode("", clusterName + "#" + address, UUIDs.randomBase64UUID(), hostName, address,
+                transportAddress, Collections
+                .emptyMap(), EnumSet.allOf(DiscoveryNode.Role.class),
+                Version.CURRENT.minimumCompatibilityVersion());
+        } else {
+            TransportAddress transportAddress = new TransportAddress(RemoteClusterAware.parseSeedAddress(address));
+            return new DiscoveryNode(clusterName + "#" + transportAddress.toString(),
+                transportAddress,
+                Version.CURRENT.minimumCompatibilityVersion());
+        }
+    }
+
     /**
      * Groups indices per cluster by splitting remote cluster-alias, index-name pairs on {@link #REMOTE_CLUSTER_INDEX_SEPARATOR}. All
      * indices per cluster are collected as a list in the returned map keyed by the cluster alias. Local indices are grouped under
@@ -138,20 +175,24 @@ public Map<String, List<String>> groupClusterIndices(String[] requestIndices, Pr
 
     protected abstract Set<String> getRemoteClusterNames();
 
+
     /**
      * Subclasses must implement this to receive information about updated cluster aliases. If the given address list is
      * empty the cluster alias is unregistered and should be removed.
      */
-    protected abstract void updateRemoteCluster(String clusterAlias, List<String> addresses);
+    protected abstract void updateRemoteCluster(String clusterAlias, List<String> addresses, String proxy);
 
     /**
      * Registers this instance to listen to updates on the cluster settings.
      */
     public void listenForUpdates(ClusterSettings clusterSettings) {
-        clusterSettings.addAffixUpdateConsumer(RemoteClusterAware.REMOTE_CLUSTERS_SEEDS, this::updateRemoteCluster,
+        clusterSettings.addAffixUpdateConsumer(RemoteClusterAware.REMOTE_CLUSTERS_PROXY,
+            RemoteClusterAware.REMOTE_CLUSTERS_SEEDS,
+            (key, value) -> updateRemoteCluster(key, value.v2(), value.v1()),
             (namespace, value) -> {});
     }
 
+
     protected static InetSocketAddress parseSeedAddress(String remoteHost) {
         String host = remoteHost.substring(0, indexOfPortSeparator(remoteHost));
         InetAddress hostAddress;