Fix _field_caps serialization in order to support cross cluster sea…

…rch (#24722)

Today the `_field_caps` API doesn't implement its request serialization
correctly since indices and indices options are not serialized at all.
This will likely break with all transport clients etc. and if this request
must be send across the network. This commit fixes this and adds correct
handling if we have only remote indices to prevent the inclusion of
all local indices.

core/src/main/java/org/elasticsearch/action/OriginalIndices.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 
 import java.io.IOException;
+import java.util.Arrays;
 
 /**
  * Used to keep track of original indices within internal (e.g. shard level) requests
@@ -64,4 +65,12 @@ public static void writeOriginalIndices(OriginalIndices originalIndices, StreamO
         out.writeStringArrayNullable(originalIndices.indices);
         originalIndices.indicesOptions.writeIndicesOptions(out);
     }
+
+    @Override
+    public String toString() {
+        return "OriginalIndices{" +
+            "indices=" + Arrays.toString(indices) +
+            ", indicesOptions=" + indicesOptions +
+            '}';
+    }
 }

core/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java

@@ -35,6 +35,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 
 import static org.elasticsearch.common.xcontent.ObjectParser.fromList;
@@ -78,6 +79,8 @@ public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         fields = in.readStringArray();
         if (in.getVersion().onOrAfter(Version.V_5_5_0_UNRELEASED)) {
+            indices = in.readStringArray();
+            indicesOptions = IndicesOptions.readIndicesOptions(in);
             mergeResults = in.readBoolean();
         } else {
             mergeResults = true;
@@ -89,6 +92,8 @@ public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeStringArray(fields);
         if (out.getVersion().onOrAfter(Version.V_5_5_0_UNRELEASED)) {
+            out.writeStringArray(indices);
+            indicesOptions.writeIndicesOptions(out);
             out.writeBoolean(mergeResults);
         }
     }
@@ -118,12 +123,12 @@ public String[] fields() {
      * The list of indices to lookup
      */
     public FieldCapabilitiesRequest indices(String... indices) {
-        this.indices = indices;
+        this.indices = Objects.requireNonNull(indices, "indices must not be null");
         return this;
     }
 
     public FieldCapabilitiesRequest indicesOptions(IndicesOptions indicesOptions) {
-        this.indicesOptions = indicesOptions;
+        this.indicesOptions = Objects.requireNonNull(indicesOptions, "indices options must not be null");
         return this;
     }
 

core/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java

@@ -26,6 +26,7 @@
 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.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.CountDown;
@@ -72,7 +73,14 @@ protected void doExecute(FieldCapabilitiesRequest request,
         final Map<String, OriginalIndices> remoteClusterIndices = remoteClusterService.groupIndices(request.indicesOptions(),
             request.indices(), idx -> indexNameExpressionResolver.hasIndexOrAlias(idx, clusterState));
         final OriginalIndices localIndices = remoteClusterIndices.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
-        final String[] concreteIndices = indexNameExpressionResolver.concreteIndexNames(clusterState, localIndices);
+        final String[] concreteIndices;
+        if (remoteClusterIndices.isEmpty() == false && localIndices.indices().length == 0) {
+            // in the case we have one or more remote indices but no local we don't expand to all local indices and just do remote
+            // indices
+            concreteIndices = Strings.EMPTY_ARRAY;
+        } else {
+            concreteIndices = indexNameExpressionResolver.concreteIndexNames(clusterState, localIndices);
+        }
         final int totalNumRequest = concreteIndices.length + remoteClusterIndices.size();
         final CountDown completionCounter = new CountDown(totalNumRequest);
         final List<FieldCapabilitiesIndexResponse> indexResponses = Collections.synchronizedList(new ArrayList<>());

core/src/test/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequestTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.fieldcaps;
 
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.test.ESTestCase;
@@ -33,10 +34,52 @@ private FieldCapabilitiesRequest randomRequest() {
         for (int i = 0; i < size; i++) {
             randomFields[i] = randomAlphaOfLengthBetween(5, 10);
         }
+
+        size = randomIntBetween(0, 20);
+        String[] randomIndices = new String[size];
+        for (int i = 0; i < size; i++) {
+            randomIndices[i] = randomAlphaOfLengthBetween(5, 10);
+        }
         request.fields(randomFields);
+        request.indices(randomIndices);
+        if (randomBoolean()) {
+            request.indicesOptions(randomBoolean() ? IndicesOptions.strictExpand() : IndicesOptions.lenientExpandOpen());
+        }
         return request;
     }
 
+    public void testEqualsAndHashcode() {
+        FieldCapabilitiesRequest request = new FieldCapabilitiesRequest();
+        request.indices("foo");
+        request.indicesOptions(IndicesOptions.lenientExpandOpen());
+        request.fields("bar");
+
+        FieldCapabilitiesRequest other = new FieldCapabilitiesRequest();
+        other.indices("foo");
+        other.indicesOptions(IndicesOptions.lenientExpandOpen());
+        other.fields("bar");
+        assertEquals(request, request);
+        assertEquals(request, other);
+        assertEquals(request.hashCode(), other.hashCode());
+
+        // change indices
+        other.indices("foo", "bar");
+        assertNotEquals(request, other);
+        other.indices("foo");
+        assertEquals(request, other);
+
+        // change fields
+        other.fields("foo", "bar");
+        assertNotEquals(request, other);
+        other.fields("bar");
+        assertEquals(request, request);
+
+        // change indices options
+        other.indicesOptions(IndicesOptions.strictExpand());
+        assertNotEquals(request, other);
+
+    }
+
     public void testFieldCapsRequestSerialization() throws IOException {
         for (int i = 0; i < 20; i++) {
             FieldCapabilitiesRequest request = randomRequest();

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/multi_cluster/30_field_caps.yml

@@ -64,3 +64,16 @@
   - match: {fields.geo.keyword.indices:                 ["my_remote_cluster:field_caps_index_3"]}
   - is_false: fields.geo.keyword.non_searchable_indices
   - is_false: fields.geo.keyword.on_aggregatable_indices
+
+  - do:
+      field_caps:
+        index: 'my_remote_cluster:some_index_that_doesnt_exist'
+        fields: [number]
+  - match: { 'fields': {} } # empty response - this index doesn't exists
+
+  - do:
+      field_caps:
+        index: 'my_remote_cluster:field_caps_index_1'
+        fields: [number]
+  - match: {fields.number.double.searchable:            true}
+  - match: {fields.number.double.aggregatable:          true}