kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix ser/deserializers of ES 6/7-7.8/7.8+, search query, aggregation, order, terms

This commit is contained in:
kezhenxu94 2021-09-15 17:49:41 +08:00 committed by GitHub
parent 5d7c82d485
commit 73b6c67344
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 383 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearchVersion.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.library.elasticsearch.requests.factory.v6.V6Request
import org.apache.skywalking.library.elasticsearch.requests.factory.v6.codec.V6Codec;
import org.apache.skywalking.library.elasticsearch.requests.factory.v7.V78RequestFactory;
import org.apache.skywalking.library.elasticsearch.requests.factory.v7.V7RequestFactory;
import org.apache.skywalking.library.elasticsearch.requests.factory.v7.codec.V78Codec;
import org.apache.skywalking.library.elasticsearch.requests.factory.v7.codec.V7Codec;
public final class ElasticSearchVersion {
@ -53,11 +54,12 @@ public final class ElasticSearchVersion {
return;
}
if (major == 7) {
codec = V7Codec.INSTANCE;
if (minor < 8) { // [7.0, 7.8)
requestFactory = new V7RequestFactory(this);
codec = V7Codec.INSTANCE;
} else { // [7.8, 8.0)
requestFactory = new V78RequestFactory(this);
codec = V78Codec.INSTANCE;
}
return;
}

5
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v6/codec/V6Codec.java
View File

@ -26,6 +26,7 @@ import java.io.InputStream;
import org.apache.skywalking.library.elasticsearch.requests.IndexRequest;
import org.apache.skywalking.library.elasticsearch.requests.UpdateRequest;
import org.apache.skywalking.library.elasticsearch.requests.factory.Codec;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
import org.apache.skywalking.library.elasticsearch.response.Mappings;
public final class V6Codec implements Codec {
@ -59,6 +60,10 @@ public final class V6Codec implements Codec {
Mappings.class,
new V6MappingsDeserializer()
)
.addDeserializer(
IndexTemplates.class,
new V6IndexTemplatesDeserializer()
)
)
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

45
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v6/codec/V6IndexTemplatesDeserializer.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.skywalking.library.elasticsearch.requests.factory.v6.codec;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplate;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
final class V6IndexTemplatesDeserializer extends JsonDeserializer<IndexTemplates> {
public static final TypeReference<Map<String, IndexTemplate>> TYPE_REFERENCE =
new TypeReference<Map<String, IndexTemplate>>() {
};
@Override
public IndexTemplates deserialize(final JsonParser p,
final DeserializationContext ctxt)
throws IOException {
final Map<String, IndexTemplate> templates = p.getCodec().readValue(p, TYPE_REFERENCE);
if (templates == null) {
return new IndexTemplates(Collections.emptyMap());
}
return new IndexTemplates(templates);
}
}

2
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v6/codec/V6MappingsDeserializer.java
View File

@ -33,7 +33,7 @@ final class V6MappingsDeserializer extends JsonDeserializer<Mappings> {
throws IOException {
final Map<String, Object> m =
p.readValueAs(new TypeReference<Map<String, Object>>() {
p.getCodec().readValue(p, new TypeReference<Map<String, Object>>() {
});
final Optional<Map.Entry<String, Object>> typeMapping =
m.entrySet()

82
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v7/codec/V78Codec.java Normal file
View File

@ -0,0 +1,82 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.skywalking.library.elasticsearch.requests.factory.v7.codec;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
import java.io.InputStream;
import org.apache.skywalking.library.elasticsearch.requests.IndexRequest;
import org.apache.skywalking.library.elasticsearch.requests.UpdateRequest;
import org.apache.skywalking.library.elasticsearch.requests.factory.Codec;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
import org.apache.skywalking.library.elasticsearch.response.Mappings;
public final class V78Codec implements Codec {
public static final Codec INSTANCE = new V78Codec();
private static final ObjectMapper MAPPER = new ObjectMapper()
.setSerializationInclusion(JsonInclude.Include.NON_NULL)
// We added some serializers here and some in their item classes as annotation (e.g.
// org.apache.skywalking.library.elasticsearch.requests.search.Sorts),
// the basic idea is, if the item class is very basic and are the same serialization method
// in both 6.x and 7.x, we set the serializer in their item class as annotation to make it
// shared by 6.x and 7.x, without duplicating the serializer codes, otherwise, we add
// serializers for each version explicitly in the object mapper.
// The 2 methods to add serializers can be changed if some day the basic serializer cannot
// be shared between newer versions of ElasticSearch or vice versa.
.registerModule(
new SimpleModule()
.addSerializer(
IndexRequest.class,
new V7IndexRequestSerializer()
)
.addSerializer(
UpdateRequest.class,
new V7UpdateRequestSerializer()
)
.addDeserializer(
Mappings.class,
new V7MappingsDeserializer()
)
.addDeserializer(
IndexTemplates.class,
new V78IndexTemplatesDeserializer()
)
)
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
@Override
public byte[] encode(final Object request) throws Exception {
return MAPPER.writeValueAsBytes(request);
}
@Override
public <T> T decode(final InputStream inputStream,
final TypeReference<T> type) throws Exception {
return MAPPER.readValue(inputStream, type);
}
@Override
public <T> T decode(final InputStream inputStream,
final Class<T> clazz) throws Exception {
return MAPPER.readValue(inputStream, clazz);
}
}

96
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v7/codec/V78IndexTemplatesDeserializer.java Normal file
View File

@ -0,0 +1,96 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.skywalking.library.elasticsearch.requests.factory.v7.codec;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.core.io.SerializedString;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Strings;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplate;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
import static java.util.stream.Collectors.toMap;
@Slf4j
final class V78IndexTemplatesDeserializer extends JsonDeserializer<IndexTemplates> {
private static final TypeReference<List<IndexTemplateWrapper>> TYPE_REFERENCE =
new TypeReference<List<IndexTemplateWrapper>>() {
};
@Override
public IndexTemplates deserialize(final JsonParser p,
final DeserializationContext ctxt)
throws IOException {
while (!p.nextFieldName(new SerializedString("index_templates"))) {
if (p.currentName() == null) {
return new IndexTemplates(Collections.emptyMap());
}
p.skipChildren();
}
if (p.nextToken() != JsonToken.START_ARRAY) {
throw new UnsupportedOperationException(
"this might be a new ElasticSearch version and we don't support yet");
}
final JsonNode array = p.getCodec().readTree(p);
final List<IndexTemplate> templates = new ArrayList<>(array.size());
for (final JsonNode node : array) {
final String name = node.get("name").asText();
if (Strings.isNullOrEmpty(name)) {
log.error("index template without a name: {}", node);
continue;
}
final JsonNode indexTemplateNode = node.get("index_template");
if (indexTemplateNode == null) {
log.error("index template without index_template: {}", node);
continue;
}
final IndexTemplateWrapper wrapper =
p.getCodec().treeToValue(indexTemplateNode, IndexTemplateWrapper.class);
wrapper.getTemplate().setName(name);
wrapper.getTemplate().setIndexPatterns(wrapper.getIndexPatterns());
templates.add(wrapper.getTemplate());
}
final Map<String, IndexTemplate> templateMap =
templates.stream()
.collect(toMap(IndexTemplate::getName, Function.identity()));
return new IndexTemplates(templateMap);
}
@Data
static final class IndexTemplateWrapper {
@JsonProperty("index_patterns")
private List<String> indexPatterns;
private IndexTemplate template;
}
}

5
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v7/codec/V7Codec.java
View File

@ -26,6 +26,7 @@ import java.io.InputStream;
import org.apache.skywalking.library.elasticsearch.requests.IndexRequest;
import org.apache.skywalking.library.elasticsearch.requests.UpdateRequest;
import org.apache.skywalking.library.elasticsearch.requests.factory.Codec;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
import org.apache.skywalking.library.elasticsearch.response.Mappings;
public final class V7Codec implements Codec {
@ -55,6 +56,10 @@ public final class V7Codec implements Codec {
Mappings.class,
new V7MappingsDeserializer()
)
.addDeserializer(
IndexTemplates.class,
new V7IndexTemplatesDeserializer()
)
)
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

46
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v7/codec/V7IndexTemplatesDeserializer.java Normal file
View File

@ -0,0 +1,46 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.skywalking.library.elasticsearch.requests.factory.v7.codec;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplate;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplates;
final class V7IndexTemplatesDeserializer extends JsonDeserializer<IndexTemplates> {
public static final TypeReference<Map<String, IndexTemplate>> TYPE_REFERENCE =
new TypeReference<Map<String, IndexTemplate>>() {
};
@Override
public IndexTemplates deserialize(final JsonParser p,
final DeserializationContext ctxt)
throws IOException {
final Map<String, IndexTemplate> templates = p.getCodec().readValue(p, TYPE_REFERENCE);
if (templates == null) {
return new IndexTemplates(Collections.emptyMap());
}
return new IndexTemplates(templates);
}
}

4
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/v7/codec/V7MappingsDeserializer.java
View File

@ -33,14 +33,14 @@ final class V7MappingsDeserializer extends JsonDeserializer<Mappings> {
throws IOException {
final Map<String, Object> m =
p.readValueAs(new TypeReference<Map<String, Object>>() {
p.getCodec().readValue(p, new TypeReference<Map<String, Object>>() {
});
final Iterator<Map.Entry<String, Object>> it = m.entrySet().iterator();
if (it.hasNext()) {
final Map.Entry<String, Object> first = it.next();
final Mappings mappings = new Mappings();
mappings.setType(first.getKey());
mappings.setType("_doc");
mappings.setProperties((Map<String, Object>) first.getValue());
return mappings;
}

63
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/BoolQueryBuilder.java
View File

@ -19,83 +19,64 @@ package org.apache.skywalking.library.elasticsearch.requests.search;
import com.google.common.collect.ImmutableList;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static java.util.Objects.requireNonNull;
public final class BoolQueryBuilder implements QueryBuilder {
private ImmutableList.Builder<Query> must;
private ImmutableList.Builder<Query> mustNot;
private ImmutableList.Builder<Query> should;
private ImmutableList.Builder<Query> shouldNot;
private ImmutableList.Builder<QueryBuilder> must;
private ImmutableList.Builder<QueryBuilder> mustNot;
private ImmutableList.Builder<QueryBuilder> should;
private ImmutableList.Builder<QueryBuilder> shouldNot;
BoolQueryBuilder() {
}
public BoolQueryBuilder must(Query query) {
requireNonNull(query, "query");
must().add(query);
return this;
}
public BoolQueryBuilder must(QueryBuilder queryBuilder) {
requireNonNull(queryBuilder, "queryBuilder");
return must(queryBuilder.build());
}
public BoolQueryBuilder mustNot(Query query) {
requireNonNull(query, "query");
mustNot().add(query);
must().add(queryBuilder);
return this;
}
public BoolQueryBuilder mustNot(QueryBuilder queryBuilder) {
requireNonNull(queryBuilder, "queryBuilder");
return mustNot(queryBuilder.build());
}
public BoolQueryBuilder should(Query query) {
requireNonNull(query, "query");
should().add(query);
mustNot().add(queryBuilder);
return this;
}
public BoolQueryBuilder should(QueryBuilder queryBuilder) {
requireNonNull(queryBuilder, "queryBuilder");
return should(queryBuilder.build());
}
public BoolQueryBuilder shouldNot(Query query) {
requireNonNull(query, "query");
shouldNot().add(query);
should().add(queryBuilder);
return this;
}
public BoolQueryBuilder shouldNot(QueryBuilder queryBuilder) {
requireNonNull(queryBuilder, "queryBuilder");
return shouldNot(queryBuilder.build());
shouldNot().add(queryBuilder);
return this;
}
private ImmutableList.Builder<Query> must() {
private ImmutableList.Builder<QueryBuilder> must() {
if (must == null) {
must = ImmutableList.builder();
}
return must;
}
private ImmutableList.Builder<Query> mustNot() {
private ImmutableList.Builder<QueryBuilder> mustNot() {
if (mustNot == null) {
mustNot = ImmutableList.builder();
}
return mustNot;
}
private ImmutableList.Builder<Query> should() {
private ImmutableList.Builder<QueryBuilder> should() {
if (should == null) {
should = ImmutableList.builder();
}
return should;
}
private ImmutableList.Builder<Query> shouldNot() {
private ImmutableList.Builder<QueryBuilder> shouldNot() {
if (shouldNot == null) {
shouldNot = ImmutableList.builder();
}
@ -108,25 +89,33 @@ public final class BoolQueryBuilder implements QueryBuilder {
if (this.must == null) {
must = null;
} else {
must = this.must.build();
must = this.must.build().stream()
.map(QueryBuilder::build)
.collect(toImmutableList());
}
final ImmutableList<Query> should;
if (this.should == null) {
should = null;
} else {
should = this.should.build();
should = this.should.build().stream()
.map(QueryBuilder::build)
.collect(toImmutableList());
}
final ImmutableList<Query> mustNot;
if (this.mustNot == null) {
mustNot = null;
} else {
mustNot = this.mustNot.build();
mustNot = this.mustNot.build().stream()
.map(QueryBuilder::build)
.collect(toImmutableList());
}
final ImmutableList<Query> shouldNot;
if (this.shouldNot == null) {
shouldNot = null;
} else {
shouldNot = this.shouldNot.build();
shouldNot = this.shouldNot.build().stream()
.map(QueryBuilder::build)
.collect(toImmutableList());
}
return new BoolQuery(must, mustNot, should, shouldNot);
}

7
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/Query.java
View File

@ -28,7 +28,12 @@ import static java.util.Objects.requireNonNull;
/**
* Represents criteria when matching documents in ElasticSearch.
*/
public abstract class Query {
public abstract class Query implements QueryBuilder {
@Override
public Query build() {
return this;
}
public static RangeQueryBuilder range(String name) {
checkArgument(!Strings.isNullOrEmpty(name), "name cannot be blank");
return new RangeQueryBuilder(name);

19
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/SearchBuilder.java
View File

@ -30,7 +30,6 @@ import static java.util.Objects.requireNonNull;
public final class SearchBuilder {
private Integer from;
private Integer size;
private Query query;
private QueryBuilder queryBuilder;
private ImmutableList.Builder<Sort> sort;
private ImmutableMap.Builder<String, Aggregation> aggregations;
@ -59,14 +58,8 @@ public final class SearchBuilder {
return this;
}
public SearchBuilder query(Query query) {
ensureQueryIsNotSet();
this.query = requireNonNull(query, "query");
return this;
}
public SearchBuilder query(QueryBuilder queryBuilder) {
ensureQueryIsNotSet();
checkState(this.queryBuilder == null, "queryBuilder is already set");
this.queryBuilder = requireNonNull(queryBuilder, "queryBuilder");
return this;
}
@ -97,9 +90,7 @@ public final class SearchBuilder {
aggregations = aggregations().build();
}
final Query query;
if (this.query != null) {
query = this.query;
} else if (queryBuilder != null) {
if (queryBuilder != null) {
query = queryBuilder.build();
} else {
query = null;
@ -123,10 +114,4 @@ public final class SearchBuilder {
}
return aggregations;
}
private void ensureQueryIsNotSet() {
final String errMsg = "query and queryBuilder can not be set simultaneously";
checkState(query == null, errMsg);
checkState(queryBuilder == null, errMsg);
}
}

19
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/aggregation/TermsAggregation.java
View File

@ -52,6 +52,12 @@ public final class TermsAggregation extends Aggregation {
gen.writeStartObject();
{
gen.writeStringField("field", value.getField());
if (value.getSize() != null) {
gen.writeNumberField("size", value.getSize());
}
if (value.getOrder() != null) {
writeOrder(value, gen);
}
}
gen.writeEndObject();
@ -61,5 +67,18 @@ public final class TermsAggregation extends Aggregation {
}
gen.writeEndObject();
}
private void writeOrder(final TermsAggregation value,
final JsonGenerator gen) throws IOException {
gen.writeFieldName("order");
gen.writeStartObject();
{
gen.writeStringField(
value.getOrder().getPath(),
value.getOrder().isAsc() ? "asc" : "desc"
);
}
gen.writeEndObject();
}
}
}

2
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/IndexTemplate.java
View File

@ -17,6 +17,7 @@
package org.apache.skywalking.library.elasticsearch.response;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.List;
import java.util.Map;
import lombok.Data;
@ -25,6 +26,7 @@ import lombok.Data;
public final class IndexTemplate {
private String name;
private int order;
@JsonProperty("index_patterns")
private List<String> indexPatterns;
private Map<String, Object> settings;
private Mappings mappings;

4
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/IndexTemplates.java
View File

@ -22,10 +22,12 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
import lombok.Data;
import lombok.RequiredArgsConstructor;
@Data
@RequiredArgsConstructor
public final class IndexTemplates implements Iterable<IndexTemplate> {
private Map<String, IndexTemplate> templates;
private final Map<String, IndexTemplate> templates;
public Optional<IndexTemplate> get(String name) {
final Map<String, IndexTemplate> templates = getTemplates();

2
oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/Mappings.java
View File

@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@ -29,6 +30,7 @@ import lombok.ToString;
@Builder
@ToString
@EqualsAndHashCode
@NoArgsConstructor // For deserialization
@AllArgsConstructor
public final class Mappings {

31
oap-server/server-library/library-elasticsearch-client/src/test/java/org/apache/skywalking/library/elasticsearch/ITElasticSearchTest.java
View File

@ -22,10 +22,12 @@ import java.util.Collection;
import java.util.List;
import java.util.Map;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.library.elasticsearch.client.TemplateClient;
import org.apache.skywalking.library.elasticsearch.requests.IndexRequest;
import org.apache.skywalking.library.elasticsearch.requests.search.Query;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.Aggregation;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplate;
import org.apache.skywalking.library.elasticsearch.response.Mappings;
import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
import org.awaitility.Duration;
@ -38,6 +40,7 @@ import org.testcontainers.elasticsearch.ElasticsearchContainer;
import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
import org.testcontainers.utility.DockerImageName;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@ -51,7 +54,7 @@ public class ITElasticSearchTest {
@Parameterized.Parameters(name = "version: {0}")
public static Collection<Object[]> versions() {
return Arrays.asList(new Object[][] {
{"6.3.2"}, {"7.4.2"}, {"7.8.0"}
{"6.3.2"}, {"7.4.2"}, {"7.8.0"}, {"7.10.2"}
});
}
@ -79,6 +82,30 @@ public class ITElasticSearchTest {
server.stop();
}
@Test
public void testTemplate() {
final String name = "test-template";
final TemplateClient templateClient = client.templates();
final ImmutableMap<String, Object> properties = ImmutableMap.of(
"metric_table", ImmutableMap.of("type", "keyword"),
"service_id", ImmutableMap.of("type", "keyword")
);
final Mappings mappings = Mappings.builder()
.type("_doc")
.properties(properties)
.build();
assertThat(templateClient.createOrUpdate(name, ImmutableMap.of(), mappings, 0))
.isTrue();
assertThat(templateClient.get(name))
.isPresent()
.map(IndexTemplate::getMappings)
.map(Mappings::getProperties)
.hasValue(mappings.getProperties());
}
@Test
public void testIndex() {
final String index = "test-index";
@ -167,7 +194,7 @@ public class ITElasticSearchTest {
.must(Query.term("key1", "val3"))
.must(Query.term("key2", "val4"))
.build()
).build()))
)))
.aggregation(
Aggregation
.terms("key1").field("key1.keyword")

9
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
View File

@ -63,7 +63,7 @@ public class StorageEsInstaller extends ModelInstaller {
}
@Override
protected boolean isExists(Model model) throws StorageException {
protected boolean isExists(Model model) {
ElasticSearchClient esClient = (ElasticSearchClient) client;
String tableName = IndexController.INSTANCE.getTableName(model);
IndexController.LogicIndicesRegister.registerRelation(model.getName(), tableName);
@ -73,7 +73,12 @@ public class StorageEsInstaller extends ModelInstaller {
boolean exist = esClient.isExistsTemplate(tableName)
&& esClient.isExistsIndex(TimeSeriesUtils.latestWriteIndexName(model));
final Optional<IndexTemplate> template = esClient.getTemplate(tableName);
if (exist && template.isPresent() && IndexController.INSTANCE.isMetricModel(model)) {
if ((exist && !template.isPresent()) || (!exist && template.isPresent())) {
throw new Error("[Bug warning]ElasticSearch client query template result is not consistent. Please file an issue to Apache SkyWalking.(https://github.com/apache/skywalking/issues)");
}
if (exist && IndexController.INSTANCE.isMetricModel(model)) {
structures.putStructure(
tableName, template.get().getMappings()
);

4
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java
View File

@ -82,7 +82,7 @@ public class AggregationQueryEsDAO extends EsDAO implements IAggregationQueryDAO
condition.getName()
));
additionalConditions.forEach(additionalCondition -> boolQuery
.must(Query.term(
.must(Query.terms(
additionalCondition.getKey(),
additionalCondition.getValue()
)));
@ -91,7 +91,7 @@ public class AggregationQueryEsDAO extends EsDAO implements IAggregationQueryDAO
} else {
final BoolQueryBuilder boolQuery = Query.bool();
additionalConditions.forEach(additionalCondition -> boolQuery
.must(Query.term(
.must(Query.terms(
additionalCondition.getKey(),
additionalCondition.getValue()
)));