Fix Rollup's metadata parser (#36791)

The parser used for rollup configs in _meta fields was not able to
handle unrelated data in the meta field.  If an unrelated object
was encountered, it would half-consume the JSON object, realize it
wasn'ta rollup config, then stop parsing.  This would leave the object
halfway consumed and the parsing framework would throw an exception.

This commit replaces the parsing logic with a set of minimal parsers,
each for the specific component we care about (`_doc`, `_meta`,
`_rollup`) and configured to ignore unknown fields where applicable.

More verbose, but less hacky than before and should be more robust.

Also adds tests (randomized and explicit) to make sure this doesn't
break in the future.

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/RollupIndexCaps.java

@@ -11,6 +11,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ObjectParser;
@@ -31,11 +32,13 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-public class RollupIndexCaps implements Writeable, ToXContentFragment {
-    static ParseField ROLLUP_JOBS = new ParseField("rollup_jobs");
-    private static ParseField INDEX_NAME = new ParseField(RollupField.TYPE_NAME);
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
 
-    //TODO find a way to make this parsing less hacky :(
+public class RollupIndexCaps implements Writeable, ToXContentFragment {
+    private static ParseField ROLLUP_JOBS = new ParseField("rollup_jobs");
+    private static ParseField DOC_FIELD = new ParseField("_doc");
+    private static ParseField META_FIELD = new ParseField("_meta");
+    private static ParseField ROLLUP_FIELD = new ParseField(RollupField.ROLLUP_META);
     // Note: we ignore unknown fields since there may be unrelated metadata
     private static final ObjectParser<RollupIndexCaps, Void> METADATA_PARSER
             = new ObjectParser<>(GetRollupCapsAction.NAME, true, RollupIndexCaps::new);
@@ -57,33 +60,55 @@ public class RollupIndexCaps implements Writeable, ToXContentFragment {
               }
             }
          */
-        METADATA_PARSER.declareField((parser, rollupIndexCaps, aVoid) -> {
-            // "_doc"
-            if (parser.currentName().equals(RollupField.TYPE_NAME) && parser.currentToken().equals(XContentParser.Token.START_OBJECT)) {
-                parser.nextToken();// START_OBJECT
-                List<RollupJobConfig> jobs = new ArrayList<>();
-
-                // "meta"
-                if (parser.currentName().equals("_meta") && parser.currentToken().equals(XContentParser.Token.FIELD_NAME)) {
-                    parser.nextToken(); // FIELD_NAME
-                    parser.nextToken(); // START_OBJECT
-
-                    // "_rollup"
-                    if (parser.currentName().equals(RollupField.ROLLUP_META) &&
-                            parser.currentToken().equals(XContentParser.Token.FIELD_NAME)) {
-                        parser.nextToken(); // FIELD_NAME
-
-                        // "job-1"
-                        while (parser.nextToken().equals(XContentParser.Token.END_OBJECT) == false) {
-                            jobs.add(RollupJobConfig.fromXContent(parser, null));
-                        }
+        METADATA_PARSER.declareField((parser, rollupIndexCaps, aVoid)
+                -> rollupIndexCaps.setJobs(DocParser.DOC_PARSER.apply(parser, aVoid).jobs),
+            DOC_FIELD, ObjectParser.ValueType.OBJECT);
+    }
+
+    /**
+     * Parser for `_doc` portion of mapping metadata
+     */
+    private static class DocParser {
+        public List<RollupJobConfig> jobs;
+        // Ignore unknown fields because there could be unrelated doc types
+        private static final ConstructingObjectParser<DocParser, Void> DOC_PARSER
+            = new ConstructingObjectParser<>("_rollup_doc_parser", true, a -> {
+                List<RollupJobConfig> j = new ArrayList<>();
+                for (Object o : (List)a[0]) {
+                    if (o instanceof RollupJobConfig) {
+                        j.add((RollupJobConfig) o);
                     }
                 }
-                rollupIndexCaps.setJobs(jobs);
-            }
-        }, INDEX_NAME, ObjectParser.ValueType.OBJECT);
+                return new DocParser(j);
+            });
+
+        static {
+            DOC_PARSER.declareField(constructorArg(), MetaParser.META_PARSER::apply, META_FIELD, ObjectParser.ValueType.OBJECT);
+        }
+
+        DocParser(List<RollupJobConfig> jobs) {
+            this.jobs = jobs;
+        }
     }
 
+    /**
+     * Parser for `_meta` portion of mapping metadata
+     */
+    private static class MetaParser {
+        // Ignore unknown fields because there could be unrelated _meta values
+        private static final ObjectParser<List<RollupJobConfig>, Void> META_PARSER
+            = new ObjectParser<>("_rollup_meta_parser", true, ArrayList::new);
+        static {
+            META_PARSER.declareField((parser, jobs, aVoid) -> {
+                // "job-1"
+                while (parser.nextToken().equals(XContentParser.Token.END_OBJECT) == false) {
+                    jobs.add(RollupJobConfig.fromXContent(parser, null));
+                }
+            }, ROLLUP_FIELD, ObjectParser.ValueType.OBJECT);
+        }
+    }
+
+
     private List<RollupJobCaps> jobCaps = Collections.emptyList();
     private String rollupIndexName;
 

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/GetRollupCapsActionRequestTests.java

@@ -198,6 +198,100 @@ public class GetRollupCapsActionRequestTests extends AbstractStreamableTestCase<
         Map<String, RollableIndexCaps> caps = TransportGetRollupCapsAction.getCaps(selectedIndexName, indices.build());
         assertThat(caps.size(), equalTo(1));
     }
+
+    public void testNonRollupMeta() throws IOException {
+        String indexPattern = randomBoolean() ? randomAlphaOfLength(10) : randomAlphaOfLength(10) + "-*";
+
+        MappingMetaData mappingMeta = new MappingMetaData(RollupField.TYPE_NAME,
+            Collections.singletonMap(RollupField.TYPE_NAME,
+                Collections.singletonMap("_meta",
+                    Collections.singletonMap("foo",
+                        Collections.singletonMap("bar", "baz")))));
+
+        ImmutableOpenMap.Builder<String, MappingMetaData> mappings = ImmutableOpenMap.builder(1);
+        mappings.put(RollupField.TYPE_NAME, mappingMeta);
+        IndexMetaData meta = Mockito.mock(IndexMetaData.class);
+        Mockito.when(meta.getMappings()).thenReturn(mappings.build());
+        Optional<RollupIndexCaps> caps = TransportGetRollupCapsAction.findRollupIndexCaps(indexPattern, meta);
+        assertFalse(caps.isPresent());
+    }
+
+    public void testNonRollupPlusRollupMeta() throws IOException {
+        String indexPattern = randomBoolean() ? randomAlphaOfLength(10) : randomAlphaOfLength(10) + "-*";
+        String jobName = randomAlphaOfLength(5);
+        RollupJobConfig job = ConfigTestHelpers.randomRollupJobConfig(random(), jobName);
+
+        Map<String, Object> metaMap = new HashMap<>(2);
+        metaMap.put("foo", Collections.singletonMap("bar", "baz"));
+        metaMap.put(RollupField.ROLLUP_META, Collections.singletonMap(jobName, job));
+
+        MappingMetaData mappingMeta = new MappingMetaData(RollupField.TYPE_NAME,
+            Collections.singletonMap(RollupField.TYPE_NAME,
+                Collections.singletonMap("_meta", metaMap)));
+
+        ImmutableOpenMap.Builder<String, MappingMetaData> mappings = ImmutableOpenMap.builder(1);
+        mappings.put(RollupField.TYPE_NAME, mappingMeta);
+        IndexMetaData meta = Mockito.mock(IndexMetaData.class);
+        Mockito.when(meta.getMappings()).thenReturn(mappings.build());
+        Optional<RollupIndexCaps> caps = TransportGetRollupCapsAction.findRollupIndexCaps(indexPattern, meta);
+        assertTrue(caps.isPresent());
+        assertThat(caps.get().getJobCaps().size(), equalTo(1));
+        assertThat(caps.get().getJobCaps().get(0).getJobID(), equalTo(jobName));
+    }
+
+    public void testRandomNonRollupPlusRollupMeta() throws IOException {
+        String indexPattern = randomBoolean() ? randomAlphaOfLength(10) : randomAlphaOfLength(10) + "-*";
+
+        Map<String, Object> metaMap = new HashMap<>();
+        int numUnrelated = randomIntBetween(0, 10);
+        for (int i = 0; i < numUnrelated; i++) {
+            int numFields = randomIntBetween(0, 5);
+            Map<String, Object> fields = new HashMap<>(numFields);
+            for (int j = 0; j < numFields; j++) {
+                int numFields2 = randomIntBetween(0, 2);
+                Map<String, String> fields2 = new HashMap<>(numFields2);
+                for (int k = 0; k < numFields; k++) {
+                    fields2.put(randomAlphaOfLength(5), randomAlphaOfLength(5));
+                }
+                fields.put(randomAlphaOfLength(5), fields2);
+            }
+            metaMap.put(randomAlphaOfLength(5), fields);
+        }
+
+        int numJobs = randomIntBetween(1,5);
+        Map<String, Object> jobs = new HashMap<>(numJobs);
+        for (int i = 0; i < numJobs; i++) {
+            String name = randomAlphaOfLength(5);
+            jobs.put(name, ConfigTestHelpers.randomRollupJobConfig(random(), name));
+        }
+        metaMap.put(RollupField.ROLLUP_META, jobs);
+
+        MappingMetaData mappingMeta = new MappingMetaData(RollupField.TYPE_NAME,
+            Collections.singletonMap(RollupField.TYPE_NAME,
+                Collections.singletonMap("_meta", metaMap)));
+
+        ImmutableOpenMap.Builder<String, MappingMetaData> mappings = ImmutableOpenMap.builder(1);
+        mappings.put(RollupField.TYPE_NAME, mappingMeta);
+        IndexMetaData meta = Mockito.mock(IndexMetaData.class);
+        Mockito.when(meta.getMappings()).thenReturn(mappings.build());
+        Optional<RollupIndexCaps> caps = TransportGetRollupCapsAction.findRollupIndexCaps(indexPattern, meta);
+        assertTrue(caps.isPresent());
+        assertThat(caps.get().getJobCaps().size(), equalTo(numJobs));
+    }
+
+    public void testEmptyType() throws IOException {
+        String indexPattern = randomBoolean() ? randomAlphaOfLength(10) : randomAlphaOfLength(10) + "-*";
+
+        MappingMetaData mappingMeta = new MappingMetaData(RollupField.TYPE_NAME,
+            Collections.singletonMap(RollupField.TYPE_NAME, Collections.emptyMap()));
+
+        ImmutableOpenMap.Builder<String, MappingMetaData> mappings = ImmutableOpenMap.builder(1);
+        mappings.put(RollupField.TYPE_NAME, mappingMeta);
+        IndexMetaData meta = Mockito.mock(IndexMetaData.class);
+        Mockito.when(meta.getMappings()).thenReturn(mappings.build());
+        Optional<RollupIndexCaps> caps = TransportGetRollupCapsAction.findRollupIndexCaps(indexPattern, meta);
+        assertFalse(caps.isPresent());
+    }
 }