Merge pull request #16145 from s1monw/feature/ingest
Fix IngestMetadata parsing and add unittest
This commit is contained in:
commit
f3c8068ba6
|
@ -223,7 +223,7 @@ public final class ObjectParser<Value, Context> implements BiFunction<XContentPa
|
|||
list.add(supplier.get()); // single value
|
||||
} else {
|
||||
while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
|
||||
if (parser.currentToken().isValue()) {
|
||||
if (parser.currentToken().isValue() || parser.currentToken() == XContentParser.Token.START_OBJECT) {
|
||||
list.add(supplier.get());
|
||||
} else {
|
||||
throw new IllegalStateException("expected value but got [" + parser.currentToken() + "]");
|
||||
|
@ -237,6 +237,11 @@ public final class ObjectParser<Value, Context> implements BiFunction<XContentPa
|
|||
declareField((p, v, c) -> consumer.accept(v, objectParser.apply(p, c)), field, ValueType.OBJECT);
|
||||
}
|
||||
|
||||
public <T> void declareObjectArray(BiConsumer<Value, List<T>> consumer, BiFunction<XContentParser, Context, T> objectParser, ParseField field) {
|
||||
declareField((p, v, c) -> consumer.accept(v, parseArray(p, () -> objectParser.apply(p, c))), field, ValueType.OBJECT_ARRAY);
|
||||
}
|
||||
|
||||
|
||||
public <T> void declareObjectOrDefault(BiConsumer<Value, T> consumer, BiFunction<XContentParser, Context, T> objectParser, Supplier<T> defaultValue, ParseField field) {
|
||||
declareField((p, v, c) -> {
|
||||
if (p.currentToken() == XContentParser.Token.VALUE_BOOLEAN) {
|
||||
|
@ -333,6 +338,7 @@ public final class ObjectParser<Value, Context> implements BiFunction<XContentPa
|
|||
INT_ARRAY(EnumSet.of(XContentParser.Token.START_ARRAY, XContentParser.Token.VALUE_NUMBER, XContentParser.Token.VALUE_STRING)),
|
||||
BOOLEAN_ARRAY(EnumSet.of(XContentParser.Token.START_ARRAY, XContentParser.Token.VALUE_BOOLEAN)),
|
||||
OBJECT(EnumSet.of(XContentParser.Token.START_OBJECT)),
|
||||
OBJECT_ARRAY(EnumSet.of(XContentParser.Token.START_OBJECT, XContentParser.Token.START_ARRAY)),
|
||||
OBJECT_OR_BOOLEAN(EnumSet.of(XContentParser.Token.START_OBJECT, XContentParser.Token.VALUE_BOOLEAN)),
|
||||
VALUE(EnumSet.of(XContentParser.Token.VALUE_BOOLEAN, XContentParser.Token.VALUE_NULL ,XContentParser.Token.VALUE_EMBEDDED_OBJECT,XContentParser.Token.VALUE_NUMBER,XContentParser.Token.VALUE_STRING));
|
||||
|
||||
|
|
|
@ -22,6 +22,7 @@ package org.elasticsearch.ingest;
|
|||
import org.elasticsearch.cluster.AbstractDiffable;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.ParsingException;
|
||||
import org.elasticsearch.common.collect.HppcMaps;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
|
@ -30,9 +31,11 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
|
|||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumSet;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
|
@ -42,7 +45,13 @@ public final class IngestMetadata extends AbstractDiffable<MetaData.Custom> impl
|
|||
|
||||
public final static String TYPE = "ingest";
|
||||
public final static IngestMetadata PROTO = new IngestMetadata();
|
||||
private final ParseField PIPELINES_FIELD = new ParseField("pipeline");
|
||||
private static final ParseField PIPELINES_FIELD = new ParseField("pipeline");
|
||||
private static final ObjectParser<List<PipelineConfiguration>, Void> INGEST_METADATA_PARSER = new ObjectParser<>("ingest_metadata", ArrayList::new);
|
||||
|
||||
static {
|
||||
INGEST_METADATA_PARSER.declareObjectArray(List::addAll , PipelineConfiguration.getParser(), PIPELINES_FIELD);
|
||||
}
|
||||
|
||||
|
||||
// We can't use Pipeline class directly in cluster state, because we don't have the processor factories around when
|
||||
// IngestMetadata is registered as custom metadata.
|
||||
|
@ -86,20 +95,11 @@ public final class IngestMetadata extends AbstractDiffable<MetaData.Custom> impl
|
|||
|
||||
@Override
|
||||
public MetaData.Custom fromXContent(XContentParser parser) throws IOException {
|
||||
ObjectParser<Void, Void> ingestMetaDataParser = new ObjectParser<>("ingest_metadata", null);
|
||||
|
||||
Map<String, PipelineConfiguration> pipelines = new HashMap<>();
|
||||
ingestMetaDataParser.declareField((parser1, aVoid, aVoid2) -> {
|
||||
XContentParser.Token token;
|
||||
while ((token = parser1.nextToken()) != XContentParser.Token.END_ARRAY) {
|
||||
if (token == XContentParser.Token.START_OBJECT) {
|
||||
PipelineConfiguration pipeline = new PipelineConfiguration.Builder(parser1).build();
|
||||
pipelines.put(pipeline.getId(), pipeline);
|
||||
}
|
||||
}
|
||||
}, PIPELINES_FIELD, ObjectParser.ValueType.OBJECT);
|
||||
ingestMetaDataParser.parse(parser);
|
||||
|
||||
List<PipelineConfiguration> configs = INGEST_METADATA_PARSER.parse(parser);
|
||||
for (PipelineConfiguration pipeline : configs) {
|
||||
pipelines.put(pipeline.getId(), pipeline);
|
||||
}
|
||||
return new IngestMetadata(pipelines);
|
||||
}
|
||||
|
||||
|
|
|
@ -35,6 +35,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
|
|||
import java.io.IOException;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
/**
|
||||
* Encapsulates a pipeline's id and configuration as a blob
|
||||
|
@ -46,36 +47,33 @@ public final class PipelineConfiguration implements Writeable<PipelineConfigurat
|
|||
public static PipelineConfiguration readPipelineConfiguration(StreamInput in) throws IOException {
|
||||
return PROTOTYPE.readFrom(in);
|
||||
}
|
||||
private final static ObjectParser<Builder, Void> PARSER = new ObjectParser<>("pipeline_config", Builder::new);
|
||||
static {
|
||||
PARSER.declareString(Builder::setId, new ParseField("id"));
|
||||
PARSER.declareField((parser, builder, aVoid) -> {
|
||||
XContentBuilder contentBuilder = XContentBuilder.builder(parser.contentType().xContent());
|
||||
XContentHelper.copyCurrentStructure(contentBuilder.generator(), parser);
|
||||
builder.setConfig(contentBuilder.bytes());
|
||||
}, new ParseField("config"), ObjectParser.ValueType.OBJECT);
|
||||
}
|
||||
|
||||
public static class Builder {
|
||||
|
||||
private final static ObjectParser<Builder, Void> PARSER = new ObjectParser<>("pipeline_config", null);
|
||||
|
||||
static {
|
||||
PARSER.declareString(Builder::setId, new ParseField("id"));
|
||||
PARSER.declareField((parser, builder, aVoid) -> {
|
||||
XContentBuilder contentBuilder = XContentBuilder.builder(parser.contentType().xContent());
|
||||
XContentHelper.copyCurrentEvent(contentBuilder.generator(), parser);
|
||||
builder.setConfig(contentBuilder.bytes());
|
||||
}, new ParseField("config"), ObjectParser.ValueType.OBJECT);
|
||||
}
|
||||
public static BiFunction<XContentParser, Void,PipelineConfiguration> getParser() {
|
||||
return (p, c) -> PARSER.apply(p ,c).build();
|
||||
}
|
||||
private static class Builder {
|
||||
|
||||
private String id;
|
||||
private BytesReference config;
|
||||
|
||||
public Builder(XContentParser parser) throws IOException {
|
||||
PARSER.parse(parser, this);
|
||||
}
|
||||
|
||||
public void setId(String id) {
|
||||
void setId(String id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public void setConfig(BytesReference config) {
|
||||
void setConfig(BytesReference config) {
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
public PipelineConfiguration build() {
|
||||
PipelineConfiguration build() {
|
||||
return new PipelineConfiguration(id, config);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,64 @@
|
|||
/*
|
||||
* 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.ingest;
|
||||
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.bytes.BytesArray;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentFactory;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.common.xcontent.XContentType;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public class IngestMetadataTests extends ESTestCase {
|
||||
|
||||
public void testFromXContent() throws IOException {
|
||||
PipelineConfiguration pipeline = new PipelineConfiguration(
|
||||
"1",new BytesArray("{\"processors\": [{\"set\" : {\"field\": \"_field\", \"value\": \"_value\"}}]}")
|
||||
);
|
||||
PipelineConfiguration pipeline2 = new PipelineConfiguration(
|
||||
"2",new BytesArray("{\"processors\": [{\"set\" : {\"field\": \"_field1\", \"value\": \"_value1\"}}]}")
|
||||
);
|
||||
Map<String, PipelineConfiguration> map = new HashMap<>();
|
||||
map.put(pipeline.getId(), pipeline);
|
||||
map.put(pipeline2.getId(), pipeline2);
|
||||
IngestMetadata ingestMetadata = new IngestMetadata(map);
|
||||
XContentBuilder builder = XContentFactory.jsonBuilder();
|
||||
builder.prettyPrint();
|
||||
builder.startObject();
|
||||
ingestMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
|
||||
builder.endObject();
|
||||
String string = builder.string();
|
||||
final XContentParser parser = XContentFactory.xContent(XContentType.JSON).createParser(string);
|
||||
MetaData.Custom custom = ingestMetadata.fromXContent(parser);
|
||||
assertTrue(custom instanceof IngestMetadata);
|
||||
IngestMetadata m = (IngestMetadata) custom;
|
||||
assertEquals(2, m.getPipelines().size());
|
||||
assertEquals("1", m.getPipelines().get("1").getId());
|
||||
assertEquals("2", m.getPipelines().get("2").getId());
|
||||
assertEquals(pipeline.getConfigAsMap(), m.getPipelines().get("1").getConfigAsMap());
|
||||
assertEquals(pipeline2.getConfigAsMap(), m.getPipelines().get("2").getConfigAsMap());
|
||||
}
|
||||
}
|
|
@ -230,7 +230,7 @@ public class CompletionSuggestSearchIT extends ESIntegTestCase {
|
|||
SuggestResponse suggestResponse = client().suggest(request).get();
|
||||
assertThat(suggestResponse.getSuccessfulShards(), equalTo(0));
|
||||
for (ShardOperationFailedException exception : suggestResponse.getShardFailures()) {
|
||||
assertThat(exception.reason(), containsString("ParsingException[[completion] failed to parse field [payload]]; nested: IllegalStateException[expected value but got [START_OBJECT]]"));
|
||||
assertThat(exception.reason(), containsString("ParsingException[[completion] failed to parse field [payload]]; nested: IllegalStateException[Can't get text on a START_OBJECT"));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue