add support for write index resolution when creating/updating documents

modules/reindex/src/test/java/org/elasticsearch/index/reindex/ReindexSourceTargetValidationTests.java

@@ -46,13 +46,15 @@
  */
 public class ReindexSourceTargetValidationTests extends ESTestCase {
     private static final ClusterState STATE = ClusterState.builder(new ClusterName("test")).metaData(MetaData.builder()
-                .put(index("target", "target_alias", "target_multi"), true)
-                .put(index("target2", "target_multi"), true)
-                .put(index("foo"), true)
-                .put(index("bar"), true)
-                .put(index("baz"), true)
-                .put(index("source", "source_multi"), true)
-                .put(index("source2", "source_multi"), true)).build();
+                .put(index("target", false, "target_alias", "target_multi"), true)
+                .put(index("target2", false, "target_multi"), true)
+                .put(index("target3", true, "target_multi_with_write"), true)
+                .put(index("target4", false, "target_multi_with_write"), true)
+                .put(index("foo", false), true)
+                .put(index("bar", false), true)
+                .put(index("baz", false), true)
+                .put(index("source", false, "source_multi"), true)
+                .put(index("source2", false, "source_multi"), true)).build();
     private static final IndexNameExpressionResolver INDEX_NAME_EXPRESSION_RESOLVER = new IndexNameExpressionResolver(Settings.EMPTY);
     private static final AutoCreateIndex AUTO_CREATE_INDEX = new AutoCreateIndex(Settings.EMPTY,
             new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), INDEX_NAME_EXPRESSION_RESOLVER);
@@ -84,6 +86,7 @@ public void testTargetIsAlias() {
         // The index names can come in either order
         assertThat(e.getMessage(), containsString("target"));
         assertThat(e.getMessage(), containsString("target2"));
+        // NORELEASE: succeeds("target_multi_with_write", "foo"); should this be allowed?
     }
 
     public void testRemoteInfoSkipsValidation() {
@@ -109,7 +112,7 @@ private void succeeds(RemoteInfo remoteInfo, String target, String... sources) {
                 INDEX_NAME_EXPRESSION_RESOLVER, AUTO_CREATE_INDEX, STATE);
     }
 
-    private static IndexMetaData index(String name, String... aliases) {
+    private static IndexMetaData index(String name, boolean isWriteIndex, String... aliases) {
         IndexMetaData.Builder builder = IndexMetaData.builder(name).settings(Settings.builder()
                 .put("index.version.created", Version.CURRENT.id)
                 .put("index.number_of_shards", 1)

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -295,7 +295,7 @@ protected void doRun() throws Exception {
                             TransportUpdateAction.resolveAndValidateRouting(metaData, concreteIndex.getName(), (UpdateRequest) docWriteRequest);
                             break;
                         case DELETE:
-                            docWriteRequest.routing(metaData.resolveIndexRouting(docWriteRequest.routing(), docWriteRequest.index()));
+                            docWriteRequest.routing(metaData.resolveIndexRouting(docWriteRequest.routing(), docWriteRequest.index(), true));
                             // check if routing is required, if so, throw error if routing wasn't specified
                             if (docWriteRequest.routing() == null && metaData.routingRequired(concreteIndex.getName(), docWriteRequest.type())) {
                                 throw new RoutingMissingException(concreteIndex.getName(), docWriteRequest.type(), docWriteRequest.id());
@@ -474,7 +474,7 @@ Index getConcreteIndex(String indexOrAlias) {
         Index resolveIfAbsent(DocWriteRequest<?> request) {
             Index concreteIndex = indices.get(request.index());
             if (concreteIndex == null) {
-                concreteIndex = indexNameExpressionResolver.concreteSingleIndex(state, request);
+                concreteIndex = indexNameExpressionResolver.concreteWriteIndex(state, request);
                 indices.put(request.index(), concreteIndex);
             }
             return concreteIndex;

server/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -69,7 +69,7 @@ protected ShardIterator shards(ClusterState state, InternalRequest request) {
     @Override
     protected void resolveRequest(ClusterState state, InternalRequest request) {
         // update the routing (request#index here is possibly an alias)
-        request.request().routing(state.metaData().resolveIndexRouting(request.request().routing(), request.request().index()));
+        request.request().routing(state.metaData().resolveIndexRouting(request.request().routing(), request.request().index(), false));
         // Fail fast on the node that received the request.
         if (request.request().routing() == null && state.getMetaData().routingRequired(request.concreteIndex(), request.request().type())) {
             throw new RoutingMissingException(request.concreteIndex(), request.request().type(), request.request().id());

server/src/main/java/org/elasticsearch/action/get/TransportMultiGetAction.java

@@ -68,7 +68,7 @@ protected void doExecute(Task task, final MultiGetRequest request, final ActionL
             try {
                 concreteSingleIndex = indexNameExpressionResolver.concreteSingleIndex(clusterState, item).getName();
 
-                item.routing(clusterState.metaData().resolveIndexRouting(item.routing(), item.index()));
+                item.routing(clusterState.metaData().resolveIndexRouting(item.routing(), item.index(), false));
                 if ((item.routing() == null) && (clusterState.getMetaData().routingRequired(concreteSingleIndex, item.type()))) {
                     String message = "routing is required for [" + concreteSingleIndex + "]/[" + item.type() + "]/[" + item.id() + "]";
                     responses.set(i, newItemFailure(concreteSingleIndex, item.type(), item.id(), new IllegalArgumentException(message)));

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -496,7 +496,7 @@ public void process(Version indexCreatedVersion, @Nullable MappingMetaData mappi
 
     /* resolve the routing if needed */
     public void resolveRouting(MetaData metaData) {
-        routing(metaData.resolveIndexRouting(routing, index));
+        routing(metaData.resolveIndexRouting(routing, index, true));
     }
 
     @Override
@@ -603,5 +603,4 @@ public long getAutoGeneratedTimestamp() {
     public IndexRequest setShardId(ShardId shardId) {
         throw new UnsupportedOperationException("shard id should never be set on IndexRequest");
     }
-
 }

server/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java

@@ -65,7 +65,7 @@ protected void doExecute(Task task, final MultiTermVectorsRequest request, final
         Map<ShardId, MultiTermVectorsShardRequest> shardRequests = new HashMap<>();
         for (int i = 0; i < request.requests.size(); i++) {
             TermVectorsRequest termVectorsRequest = request.requests.get(i);
-            termVectorsRequest.routing(clusterState.metaData().resolveIndexRouting(termVectorsRequest.routing(), termVectorsRequest.index()));
+            termVectorsRequest.routing(clusterState.metaData().resolveIndexRouting(termVectorsRequest.routing(), termVectorsRequest.index(), false));
             if (!clusterState.metaData().hasConcreteIndex(termVectorsRequest.index())) {
                 responses.set(i, new MultiTermVectorsItemResponse(null, new MultiTermVectorsResponse.Failure(termVectorsRequest.index(),
                         termVectorsRequest.type(), termVectorsRequest.id(), new IndexNotFoundException(termVectorsRequest.index()))));

server/src/main/java/org/elasticsearch/action/termvectors/TransportTermVectorsAction.java

@@ -78,7 +78,7 @@ protected boolean resolveIndex(TermVectorsRequest request) {
     @Override
     protected void resolveRequest(ClusterState state, InternalRequest request) {
         // update the routing (request#index here is possibly an alias or a parent)
-        request.request().routing(state.metaData().resolveIndexRouting(request.request().routing(), request.request().index()));
+        request.request().routing(state.metaData().resolveIndexRouting(request.request().routing(), request.request().index(), false));
         // Fail fast on the node that received the request.
         if (request.request().routing() == null && state.getMetaData().routingRequired(request.concreteIndex(), request.request().type())) {
             throw new RoutingMissingException(request.concreteIndex(), request.request().type(), request.request().id());

server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java

@@ -104,7 +104,7 @@ protected void resolveRequest(ClusterState state, UpdateRequest request) {
     }
 
     public static void resolveAndValidateRouting(MetaData metaData, String concreteIndex, UpdateRequest request) {
-        request.routing((metaData.resolveIndexRouting(request.routing(), request.index())));
+        request.routing((metaData.resolveIndexRouting(request.routing(), request.index(), true)));
         // Fail fast on the node that received the request, rather than failing when translating on the index or delete request.
         if (request.routing() == null && metaData.routingRequired(concreteIndex, request.type())) {
             throw new RoutingMissingException(concreteIndex, request.type(), request.id());

server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java

@@ -103,7 +103,7 @@ public String[] concreteIndexNames(ClusterState state, IndicesOptions options, S
         return concreteIndexNames(context, indexExpressions);
     }
 
-     /**
+    /**
      * Translates the provided index expression into actual concrete indices, properly deduplicated.
      *
      * @param state             the cluster state containing all the data to resolve to expressions to concrete indices
@@ -117,7 +117,26 @@ public String[] concreteIndexNames(ClusterState state, IndicesOptions options, S
      * indices options in the context don't allow such a case.
      */
     public Index[] concreteIndices(ClusterState state, IndicesOptions options, String... indexExpressions) {
-        Context context = new Context(state, options);
+        Context context = new Context(state, options, false, false);
+        return concreteIndices(context, indexExpressions);
+    }
+
+     /**
+     * Translates the provided index expression into actual concrete indices, properly deduplicated.
+     *
+     * @param state                 the cluster state containing all the data to resolve to expressions to concrete indices
+     * @param options               defines how the aliases or indices need to be resolved to concrete indices
+     * @param resolveToWriteIndex   defines whether to require that aliases resolve to their respective write indices
+     * @param indexExpressions      expressions that can be resolved to alias or index names.
+     * @return the resolved concrete indices based on the cluster state, indices options and index expressions
+     * @throws IndexNotFoundException if one of the index expressions is pointing to a missing index or alias and the
+     * provided indices options in the context don't allow such a case, or if the final result of the indices resolution
+     * contains no indices and the indices options in the context don't allow such a case.
+     * @throws IllegalArgumentException if one of the aliases resolve to multiple indices and the provided
+     * indices options in the context don't allow such a case.
+     */
+    public Index[] concreteIndices(ClusterState state, IndicesOptions options, boolean resolveToWriteIndex, String... indexExpressions) {
+        Context context = new Context(state, options, false, resolveToWriteIndex);
         return concreteIndices(context, indexExpressions);
     }
 
@@ -194,29 +213,43 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
             }
 
             Collection<IndexMetaData> resolvedIndices = aliasOrIndex.getIndices();
-            if (resolvedIndices.size() > 1 && !options.allowAliasesToMultipleIndices()) {
+
+            if (aliasOrIndex.isAlias() && context.isResolveToWriteIndex()) {
+                AliasOrIndex.Alias alias = (AliasOrIndex.Alias) aliasOrIndex;
+                IndexMetaData writeIndex = alias.getWriteIndex();
+                if (writeIndex == null) {
+                    if (alias.getIndices().size() > 1) {
+                        throw new IllegalArgumentException("Alias [" + alias.getAliasName() +
+                            "] points to multiple indices with none set as a write-index [is_write_index=true]");
+                    } else {
+                        throw new IllegalArgumentException("Alias [" + alias.getAliasName() + "] points to an index ["
+                            + alias.getIndices().get(0).getIndex().getName() + "] with [is_write_index=false]");
+                    }
+                }
+                concreteIndices.add(writeIndex.getIndex());
+            } else if (resolvedIndices.size() > 1 && options.allowAliasesToMultipleIndices() == false) {
                 String[] indexNames = new String[resolvedIndices.size()];
                 int i = 0;
                 for (IndexMetaData indexMetaData : resolvedIndices) {
                     indexNames[i++] = indexMetaData.getIndex().getName();
                 }
                 throw new IllegalArgumentException("Alias [" + expression + "] has more than one indices associated with it [" +
-                        Arrays.toString(indexNames) + "], can't execute a single index op");
-            }
-
-            for (IndexMetaData index : resolvedIndices) {
-                if (index.getState() == IndexMetaData.State.CLOSE) {
-                    if (failClosed) {
-                        throw new IndexClosedException(index.getIndex());
-                    } else {
-                        if (options.forbidClosedIndices() == false) {
-                            concreteIndices.add(index.getIndex());
+                    Arrays.toString(indexNames) + "], can't execute a single index op");
+            } else {
+                for (IndexMetaData index : resolvedIndices) {
+                    if (index.getState() == IndexMetaData.State.CLOSE) {
+                        if (failClosed) {
+                            throw new IndexClosedException(index.getIndex());
+                        } else {
+                            if (options.forbidClosedIndices() == false) {
+                                concreteIndices.add(index.getIndex());
+                            }
                         }
+                    } else if (index.getState() == IndexMetaData.State.OPEN) {
+                        concreteIndices.add(index.getIndex());
+                    } else {
+                        throw new IllegalStateException("index state [" + index.getState() + "] not supported");
                     }
-                } else if (index.getState() == IndexMetaData.State.OPEN) {
-                    concreteIndices.add(index.getIndex());
-                } else {
-                    throw new IllegalStateException("index state [" + index.getState() + "] not supported");
                 }
             }
         }
@@ -255,6 +288,24 @@ public Index concreteSingleIndex(ClusterState state, IndicesRequest request) {
         return indices[0];
     }
 
+    /**
+     * Utility method that allows to resolve an index expression to its corresponding single write index.
+     *
+     * @param state             the cluster state containing all the data to resolve to expression to a concrete index
+     * @param request           The request that defines how the an alias or an index need to be resolved to a concrete index
+     *                          and the expression that can be resolved to an alias or an index name.
+     * @throws IllegalArgumentException if the index resolution does not lead to an index, or leads to more than one index
+     * @return the write index obtained as a result of the index resolution
+     */
+    public Index concreteWriteIndex(ClusterState state, IndicesRequest request) {
+        String indexExpression = request.indices() != null && request.indices().length > 0 ? request.indices()[0] : null;
+        Index[] indices = concreteIndices(state, request.indicesOptions(), true, indexExpression);
+        if (indices.length != 1) {
+            throw new IllegalArgumentException("unable to return a single index as the index and options provided got resolved to multiple indices");
+        }
+        return indices[0];
+    }
+
     /**
      * @return whether the specified alias or index exists. If the alias or index contains datemath then that is resolved too.
      */
@@ -292,7 +343,7 @@ public String[] indexAliases(ClusterState state, String index, Predicate<AliasMe
                                  String... expressions) {
         // expand the aliases wildcard
         List<String> resolvedExpressions = expressions != null ? Arrays.asList(expressions) : Collections.emptyList();
-        Context context = new Context(state, IndicesOptions.lenientExpandOpen(), true);
+        Context context = new Context(state, IndicesOptions.lenientExpandOpen(), true, false);
         for (ExpressionResolver expressionResolver : expressionResolvers) {
             resolvedExpressions = expressionResolver.resolve(context, resolvedExpressions);
         }
@@ -512,24 +563,26 @@ static final class Context {
         private final IndicesOptions options;
         private final long startTime;
         private final boolean preserveAliases;
+        private final boolean resolveToWriteIndex;
 
         Context(ClusterState state, IndicesOptions options) {
             this(state, options, System.currentTimeMillis());
         }
 
-        Context(ClusterState state, IndicesOptions options, boolean preserveAliases) {
-            this(state, options, System.currentTimeMillis(), preserveAliases);
+        Context(ClusterState state, IndicesOptions options, boolean preserveAliases, boolean resolveToWriteIndex) {
+            this(state, options, System.currentTimeMillis(), preserveAliases, resolveToWriteIndex);
         }
 
         Context(ClusterState state, IndicesOptions options, long startTime) {
-           this(state, options, startTime, false);
+           this(state, options, startTime, false, false);
         }
 
-        Context(ClusterState state, IndicesOptions options, long startTime, boolean preserveAliases) {
+        Context(ClusterState state, IndicesOptions options, long startTime, boolean preserveAliases, boolean resolveToWriteIndex) {
             this.state = state;
             this.options = options;
             this.startTime = startTime;
             this.preserveAliases = preserveAliases;
+            this.resolveToWriteIndex = resolveToWriteIndex;
         }
 
         public ClusterState getState() {
@@ -552,6 +605,14 @@ public long getStartTime() {
         boolean isPreserveAliases() {
             return preserveAliases;
         }
+
+        /**
+         * This is used to require that aliases resolve to their write-index. It is currently not used in conjunction
+         * with <code>preserveAliases</code>.
+         */
+        boolean isResolveToWriteIndex() {
+            return resolveToWriteIndex;
+        }
     }
 
     private interface ExpressionResolver {