Extract reindexing logic from transport action (#46033)

This commit extracts the reindexing logic from the transport action so
that it can be incorporated into the persistent reindex work without
requiring the usage of the client.

Author: Tim-Brooks

modules/reindex/src/main/java/org/elasticsearch/index/reindex/AbstractAsyncBulkByScrollAction.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.client.ParentTaskAssigningClient;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
@@ -90,8 +91,9 @@ public abstract class AbstractAsyncBulkByScrollAction<Request extends AbstractBu
     protected final BulkByScrollTask task;
     protected final WorkerBulkByScrollTaskState worker;
     protected final ThreadPool threadPool;
+    protected final ScriptService scriptService;
+    protected final ReindexSslConfig sslConfig;
 
-    protected final Action mainAction;
     /**
      * The request for this action. Named mainRequest because we create lots of <code>request</code> variables all representing child
      * requests of this mainRequest.
@@ -114,12 +116,13 @@ public abstract class AbstractAsyncBulkByScrollAction<Request extends AbstractBu
     private final BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> scriptApplier;
     private int lastBatchSize;
 
-    public AbstractAsyncBulkByScrollAction(BulkByScrollTask task, boolean needsSourceDocumentVersions,
-                                           boolean needsSourceDocumentSeqNoAndPrimaryTerm, Logger logger, ParentTaskAssigningClient client,
-                                           ThreadPool threadPool, Action mainAction, Request mainRequest, 
-                                           ActionListener<BulkByScrollResponse> listener) {
-
+    AbstractAsyncBulkByScrollAction(BulkByScrollTask task, boolean needsSourceDocumentVersions,
+                                    boolean needsSourceDocumentSeqNoAndPrimaryTerm, Logger logger, ParentTaskAssigningClient client,
+                                    ThreadPool threadPool, Request mainRequest, ActionListener<BulkByScrollResponse> listener,
+                                    @Nullable ScriptService scriptService, @Nullable ReindexSslConfig sslConfig) {
         this.task = task;
+        this.scriptService = scriptService;
+        this.sslConfig = sslConfig;
         if (!task.isWorker()) {
             throw new IllegalArgumentException("Given task [" + task.getId() + "] must have a child worker");
         }
@@ -128,7 +131,6 @@ public AbstractAsyncBulkByScrollAction(BulkByScrollTask task, boolean needsSourc
         this.logger = logger;
         this.client = client;
         this.threadPool = threadPool;
-        this.mainAction = mainAction;
         this.mainRequest = mainRequest;
         this.listener = listener;
         BackoffPolicy backoffPolicy = buildBackoffPolicy();
@@ -158,7 +160,7 @@ public BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>>
         // The default script applier executes a no-op
         return (request, searchHit) -> request;
     }
-    
+
     /**
      * Build the {@link RequestWrapper} for a single search hit. This shouldn't handle
      * metadata or scripting. That will be handled by copyMetadata and

modules/reindex/src/main/java/org/elasticsearch/index/reindex/AsyncDeleteByQueryAction.java

@@ -32,9 +32,9 @@
 public class AsyncDeleteByQueryAction extends AbstractAsyncBulkByScrollAction<DeleteByQueryRequest, TransportDeleteByQueryAction> {
 
     public AsyncDeleteByQueryAction(BulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
-                                    ThreadPool threadPool, TransportDeleteByQueryAction action, DeleteByQueryRequest request,
-                                    ScriptService scriptService, ActionListener<BulkByScrollResponse> listener) {
-        super(task, false, true, logger, client, threadPool, action, request, listener);
+                                    ThreadPool threadPool, DeleteByQueryRequest request, ScriptService scriptService,
+                                    ActionListener<BulkByScrollResponse> listener) {
+        super(task, false, true, logger, client, threadPool, request, listener, scriptService, null);
     }
 
     @Override

modules/reindex/src/main/java/org/elasticsearch/index/reindex/ReindexValidator.java

@@ -0,0 +1,121 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.reindex;
+
+import org.apache.lucene.util.automaton.Automata;
+import org.apache.lucene.util.automaton.Automaton;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
+import org.apache.lucene.util.automaton.MinimizationOperations;
+import org.apache.lucene.util.automaton.Operations;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.support.AutoCreateIndex;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.settings.Settings;
+
+import java.util.List;
+
+class ReindexValidator {
+
+    private final CharacterRunAutomaton remoteWhitelist;
+    private final ClusterService clusterService;
+    private final IndexNameExpressionResolver resolver;
+    private final AutoCreateIndex autoCreateIndex;
+
+    ReindexValidator(Settings settings, ClusterService clusterService, IndexNameExpressionResolver resolver,
+                     AutoCreateIndex autoCreateIndex) {
+        this.remoteWhitelist = buildRemoteWhitelist(TransportReindexAction.REMOTE_CLUSTER_WHITELIST.get(settings));
+        this.clusterService = clusterService;
+        this.resolver = resolver;
+        this.autoCreateIndex = autoCreateIndex;
+    }
+
+    void initialValidation(ReindexRequest request) {
+        checkRemoteWhitelist(remoteWhitelist, request.getRemoteInfo());
+        ClusterState state = clusterService.state();
+        validateAgainstAliases(request.getSearchRequest(), request.getDestination(), request.getRemoteInfo(), resolver, autoCreateIndex,
+            state);
+    }
+
+    static void checkRemoteWhitelist(CharacterRunAutomaton whitelist, RemoteInfo remoteInfo) {
+        if (remoteInfo == null) {
+            return;
+        }
+        String check = remoteInfo.getHost() + ':' + remoteInfo.getPort();
+        if (whitelist.run(check)) {
+            return;
+        }
+        String whiteListKey = TransportReindexAction.REMOTE_CLUSTER_WHITELIST.getKey();
+        throw new IllegalArgumentException('[' + check + "] not whitelisted in " + whiteListKey);
+    }
+
+    /**
+     * Build the {@link CharacterRunAutomaton} that represents the reindex-from-remote whitelist and make sure that it doesn't whitelist
+     * the world.
+     */
+    static CharacterRunAutomaton buildRemoteWhitelist(List<String> whitelist) {
+        if (whitelist.isEmpty()) {
+            return new CharacterRunAutomaton(Automata.makeEmpty());
+        }
+        Automaton automaton = Regex.simpleMatchToAutomaton(whitelist.toArray(Strings.EMPTY_ARRAY));
+        automaton = MinimizationOperations.minimize(automaton, Operations.DEFAULT_MAX_DETERMINIZED_STATES);
+        if (Operations.isTotal(automaton)) {
+            throw new IllegalArgumentException("Refusing to start because whitelist " + whitelist + " accepts all addresses. "
+                + "This would allow users to reindex-from-remote any URL they like effectively having Elasticsearch make HTTP GETs "
+                + "for them.");
+        }
+        return new CharacterRunAutomaton(automaton);
+    }
+
+    /**
+     * Throws an ActionRequestValidationException if the request tries to index
+     * back into the same index or into an index that points to two indexes.
+     * This cannot be done during request validation because the cluster state
+     * isn't available then. Package private for testing.
+     */
+    static void validateAgainstAliases(SearchRequest source, IndexRequest destination, RemoteInfo remoteInfo,
+                                       IndexNameExpressionResolver indexNameExpressionResolver, AutoCreateIndex autoCreateIndex,
+                                       ClusterState clusterState) {
+        if (remoteInfo != null) {
+            return;
+        }
+        String target = destination.index();
+        if (false == autoCreateIndex.shouldAutoCreate(target, clusterState)) {
+            /*
+             * If we're going to autocreate the index we don't need to resolve
+             * it. This is the same sort of dance that TransportIndexRequest
+             * uses to decide to autocreate the index.
+             */
+            target = indexNameExpressionResolver.concreteWriteIndex(clusterState, destination).getName();
+        }
+        for (String sourceIndex : indexNameExpressionResolver.concreteIndexNames(clusterState, source)) {
+            if (sourceIndex.equals(target)) {
+                ActionRequestValidationException e = new ActionRequestValidationException();
+                e.addValidationError("reindex cannot write into an index its reading from [" + target + ']');
+                throw e;
+            }
+        }
+    }
+}