重构es,优化索引管理,增加按时间分表

This commit is contained in:
zhouhao 2020-03-05 17:27:29 +08:00
parent c352f9fc28
commit 6c082000fa
75 changed files with 1383 additions and 2021 deletions

View File

@ -8,22 +8,35 @@ import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.core.param.TermType;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.aggreation.bucket.Bucket;
import org.jetlinks.community.elastic.search.aggreation.bucket.BucketAggregationsStructure;
import org.jetlinks.community.elastic.search.aggreation.bucket.BucketResponse;
import org.jetlinks.community.elastic.search.aggreation.bucket.Sort;
import org.jetlinks.community.elastic.search.aggreation.enums.BucketType;
import org.jetlinks.community.elastic.search.aggreation.enums.MetricsType;
import org.jetlinks.community.elastic.search.aggreation.enums.OrderType;
import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsAggregationStructure;
import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsResponse;
import org.jetlinks.community.elastic.search.index.ElasticIndex;
import org.jetlinks.community.elastic.search.parser.QueryParamTranslateService;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.elastic.search.service.AggregationService;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
import org.jetlinks.community.timeseries.query.AggregationQueryParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;
/**
* @author bsetfeng
* @since 1.0
@ -32,28 +45,22 @@ import reactor.core.publisher.MonoSink;
@Slf4j
public class DefaultAggregationService implements AggregationService {
private final QueryParamTranslateService translateService;
private final ElasticRestClient restClient;
private final IndexOperationService indexOperationService;
private final ElasticSearchIndexManager indexManager;
@Autowired
public DefaultAggregationService(IndexOperationService indexOperationService,
ElasticRestClient restClient,
QueryParamTranslateService translateService) {
this.indexOperationService = indexOperationService;
public DefaultAggregationService(ElasticSearchIndexManager indexManager,
ElasticRestClient restClient) {
this.restClient = restClient;
this.translateService = translateService;
this.indexManager = indexManager;
}
@Override
public Mono<MetricsResponse> metricsAggregation(QueryParam queryParam,
MetricsAggregationStructure structure,
ElasticIndex provider) {
return searchSourceBuilderMono(queryParam, provider)
.map(builder -> new SearchRequest(provider.getStandardIndex())
public Mono<MetricsResponse> metricsAggregation(String index, QueryParam queryParam,
MetricsAggregationStructure structure) {
return createSearchSourceBuilder(queryParam, index)
.map(builder -> new SearchRequest(index)
.source(builder.aggregation(structure.getType().aggregationBuilder(structure.getName(), structure.getField()))))
.flatMap(request -> Mono.<SearchResponse>create(monoSink ->
restClient.getQueryClient().searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(monoSink))))
@ -61,12 +68,15 @@ public class DefaultAggregationService implements AggregationService {
}
@Override
public Mono<BucketResponse> bucketAggregation(QueryParam queryParam, BucketAggregationsStructure structure, ElasticIndex provider) {
return searchSourceBuilderMono(queryParam, provider)
.map(builder -> new SearchRequest(provider.getStandardIndex())
public Mono<BucketResponse> bucketAggregation(String index, QueryParam queryParam, BucketAggregationsStructure structure) {
return createSearchSourceBuilder(queryParam, index)
.map(builder -> new SearchRequest(index)
.source(builder.aggregation(structure.getType().aggregationBuilder(structure))))
.doOnNext(searchRequest ->
log.debug("聚合查询index:{},参数:{}", provider.getStandardIndex(), JSON.toJSON(searchRequest.source().toString())))
.doOnNext(searchRequest -> {
if (log.isDebugEnabled()) {
log.debug("聚合查询ElasticSearch:{},参数:{}", index, JSON.toJSON(searchRequest.source().toString()));
}
})
.flatMap(request -> Mono.<SearchResponse>create(monoSink ->
restClient
.getQueryClient()
@ -79,13 +89,11 @@ public class DefaultAggregationService implements AggregationService {
}
private Mono<SearchSourceBuilder> searchSourceBuilderMono(QueryParam queryParam, ElasticIndex provider) {
QueryParam tempQueryParam = queryParam.clone();
tempQueryParam.setPaging(false);
return indexOperationService.getIndexMappingMetadata(provider.getStandardIndex())
.map(metadata -> translateService.translate(tempQueryParam, metadata))
.doOnError(e -> log.error("解析queryParam错误, index:{}", provider.getStandardIndex(), e));
// return Mono.just(translateService.translate(queryParam, IndexMappingMetadata.getInstance(provider.getStandardIndex())));
private Mono<SearchSourceBuilder> createSearchSourceBuilder(QueryParam queryParam, String index) {
return indexManager.getIndexMetadata(index)
.map(metadata -> ElasticSearchConverter.convertSearchSourceBuilder(queryParam, metadata))
.doOnError(e -> log.error("解析queryParam错误:{}", index, e));
}
private <T> ActionListener<T> translatorActionListener(MonoSink<T> sink) {
@ -109,4 +117,164 @@ public class DefaultAggregationService implements AggregationService {
}
};
}
@Override
public Flux<Map<String, Object>> aggregation(String index, AggregationQueryParam aggregationQueryParam) {
QueryParam queryParam = prepareQueryParam(aggregationQueryParam);
BucketAggregationsStructure structure = createAggParameter(aggregationQueryParam);
return indexManager
.getIndexStrategy(index)
.flatMap(strategy ->
createSearchSourceBuilder(queryParam, index)
.map(builder ->
new SearchRequest(strategy.getIndexForSearch(index))
.source(builder.aggregation(structure.getType().aggregationBuilder(structure)))))
.doOnNext(searchRequest -> {
if (log.isDebugEnabled()) {
log.debug("聚合查询ElasticSearch:{},参数:{}", index, JSON.toJSON(searchRequest.source().toString()));
}
})
.flatMap(searchRequest ->
ReactorActionListener
.<SearchResponse>mono(listener ->
restClient.getQueryClient()
.searchAsync(searchRequest, RequestOptions.DEFAULT, listener)
))
.map(response -> BucketResponse.builder()
.name(structure.getName())
.buckets(structure.getType().convert(response.getAggregations().get(structure.getName())))
.build())
.flatMapIterable(BucketsParser::convert)
.take(aggregationQueryParam.getLimit())
;
}
static class BucketsParser {
private List<Map<String, Object>> result = new ArrayList<>();
public static List<Map<String, Object>> convert(BucketResponse response) {
return new BucketsParser(response).result;
}
public BucketsParser(BucketResponse response) {
this(response.getBuckets());
}
public BucketsParser(List<Bucket> buckets) {
buckets.forEach(bucket -> parser(bucket, new HashMap<>()));
}
public void parser(Bucket bucket, Map<String, Object> fMap) {
addBucketProperty(bucket, fMap);
if (bucket.getBuckets() != null && !bucket.getBuckets().isEmpty()) {
bucket.getBuckets().forEach(b -> {
Map<String, Object> map = new HashMap<>(fMap);
addBucketProperty(b, map);
parser(b, map);
});
} else {
result.add(fMap);
}
}
private void addBucketProperty(Bucket bucket, Map<String, Object> fMap) {
fMap.put(bucket.getName(), bucket.getKey());
fMap.putAll(bucket.toMap());
}
}
protected static QueryParam prepareQueryParam(AggregationQueryParam param) {
QueryParam queryParam = param.getQueryParam().clone();
queryParam.setPaging(false);
queryParam.and(param.getTimeProperty(), TermType.btw, Arrays.asList(calculateStartWithTime(param), param.getEndWithTime()));
if (queryParam.getSorts().isEmpty()) {
queryParam.orderBy(param.getTimeProperty()).desc();
}
return queryParam;
}
protected BucketAggregationsStructure createAggParameter(AggregationQueryParam param) {
List<BucketAggregationsStructure> structures = new ArrayList<>();
if (param.getGroupByTime() != null) {
structures.add(convertAggGroupTimeStructure(param));
}
if (param.getGroupBy() != null && !param.getGroupBy().isEmpty()) {
structures.addAll(getTermTypeStructures(param));
}
for (int i = 0, size = structures.size(); i < size; i++) {
if (i < size - 1) {
structures.get(i).setSubBucketAggregation(Collections.singletonList(structures.get(i + 1)));
}
if (i == size - 1) {
structures.get(i)
.setSubMetricsAggregation(param
.getAggColumns()
.stream()
.map(agg -> {
MetricsAggregationStructure metricsAggregationStructure = new MetricsAggregationStructure();
metricsAggregationStructure.setField(agg.getProperty());
metricsAggregationStructure.setName(agg.getAlias());
metricsAggregationStructure.setType(MetricsType.of(agg.getAggregation().name()));
return metricsAggregationStructure;
}).collect(Collectors.toList()));
}
}
return structures.get(0);
}
protected BucketAggregationsStructure convertAggGroupTimeStructure(AggregationQueryParam param) {
BucketAggregationsStructure structure = new BucketAggregationsStructure();
structure.setInterval(durationFormat(param.getGroupByTime().getInterval()));
structure.setType(BucketType.DATE_HISTOGRAM);
structure.setFormat(param.getGroupByTime().getFormat());
structure.setName(param.getGroupByTime().getAlias());
structure.setField(param.getGroupByTime().getProperty());
structure.setSort(Sort.desc(OrderType.KEY));
structure.setExtendedBounds(getExtendedBounds(param));
return structure;
}
protected static ExtendedBounds getExtendedBounds(AggregationQueryParam param) {
return new ExtendedBounds(calculateStartWithTime(param), param.getEndWithTime());
}
private static long calculateStartWithTime(AggregationQueryParam param) {
long startWithParam = param.getStartWithTime();
if (param.getGroupByTime() != null && param.getGroupByTime().getInterval() != null) {
long timeInterval = param.getGroupByTime().getInterval().toMillis() * param.getLimit();
long tempStartWithParam = param.getEndWithTime() - timeInterval;
startWithParam = Math.max(tempStartWithParam, startWithParam);
}
return startWithParam;
}
protected List<BucketAggregationsStructure> getTermTypeStructures(AggregationQueryParam param) {
return param.getGroupBy()
.stream()
.map(group -> {
BucketAggregationsStructure structure = new BucketAggregationsStructure();
structure.setType(BucketType.TERMS);
structure.setSize(param.getLimit());
structure.setField(group.getProperty());
structure.setName(group.getAlias());
return structure;
}).collect(Collectors.toList());
}
protected static String durationFormat(Duration duration) {
String durationStr = duration.toString();
if (durationStr.contains("S")) {
return duration.toMillis() / 1000 + "s";
} else if (!durationStr.contains("S") && durationStr.contains("M")) {
return duration.toMinutes() + "m";
} else if (!durationStr.contains("S") && !durationStr.contains("M")) {
if (duration.toHours() % 24 == 0) {
return duration.toDays() + "d";
} else {
return duration.toHours() + "h";
}
}
throw new UnsupportedOperationException("不支持的时间周期:" + duration.toString());
}
}

View File

@ -188,7 +188,4 @@ public enum BucketType {
mapping.put(OrderBuilder.of("desc", OrderType.KEY), BucketOrder.key(false));
}
public static void main(String[] args) {
System.out.println(mapping.get(OrderBuilder.of("desc", OrderType.KEY)).toString());
}
}

View File

@ -21,15 +21,11 @@ public class ElasticSearchConfiguration {
@Autowired
private ElasticSearchProperties properties;
@Bean
public ElasticRestClient elasticRestClient() {
RestHighLevelClient queryClient = new RestHighLevelClient(RestClient.builder(properties.createHosts())
.setRequestConfigCallback(properties::applyRequestConfigBuilder)
.setHttpClientConfigCallback(properties::applyHttpAsyncClientBuilder));
RestHighLevelClient writeClient = new RestHighLevelClient(RestClient.builder(properties.createHosts())
.setRequestConfigCallback(properties::applyRequestConfigBuilder)
.setHttpClientConfigCallback(properties::applyHttpAsyncClientBuilder));
return new ElasticRestClient(queryClient, writeClient);
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(properties.createHosts())
.setRequestConfigCallback(properties::applyRequestConfigBuilder)
.setHttpClientConfigCallback(properties::applyHttpAsyncClientBuilder));
return new ElasticRestClient(client, client);
}
}

View File

@ -4,6 +4,7 @@ import lombok.AllArgsConstructor;
import lombok.Getter;
import org.hswebframework.web.dict.EnumDict;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
@ -14,7 +15,7 @@ import java.util.stream.Collectors;
**/
@Getter
@AllArgsConstructor
public enum FieldDateFormat implements EnumDict<String> {
public enum ElasticDateFormat implements EnumDict<String> {
epoch_millis("epoch_millis", "毫秒"),
epoch_second("epoch_second", ""),
@ -29,9 +30,13 @@ public enum FieldDateFormat implements EnumDict<String> {
private String text;
public static String getFormat(List<FieldDateFormat> dateFormats) {
public static String getFormat(ElasticDateFormat... dateFormats) {
return getFormat(Arrays.asList(dateFormats));
}
public static String getFormat(List<ElasticDateFormat> dateFormats) {
return getFormatStr(dateFormats.stream()
.map(FieldDateFormat::getValue)
.map(ElasticDateFormat::getValue)
.collect(Collectors.toList())
);
}

View File

@ -0,0 +1,66 @@
package org.jetlinks.community.elastic.search.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.hswebframework.web.dict.EnumDict;
import org.hswebframework.web.exception.NotFoundException;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.types.*;
import org.springframework.util.StringUtils;
import java.util.function.Supplier;
@AllArgsConstructor
public enum ElasticPropertyType implements EnumDict<String> {
TEXT("text", "text", StringType::new),
BYTE("byte", "byte", () -> new IntType().min(Byte.MIN_VALUE).max(Byte.MAX_VALUE)),
SHORT("short", "short", () -> new IntType().min(Short.MIN_VALUE).max(Short.MAX_VALUE)),
INTEGER("int", "integer", IntType::new),
LONG("long", "long", LongType::new),
DATE("date", "date", DateTimeType::new),
HALF_FLOAT("half_float", "half_float", FloatType::new),
FLOAT("float", "float", FloatType::new),
DOUBLE("double", "double", DoubleType::new),
BOOLEAN("boolean", "boolean", BooleanType::new),
OBJECT("object", "object", ObjectType::new),
AUTO("auto", "auto", () -> null),
NESTED("nested", "nested", ObjectType::new),
IP("ip", "ip", LongType::new),
ATTACHMENT("attachment", "attachment", FileType::new),
KEYWORD("string", "keyword", StringType::new),
GEO_POINT("geo_point", "geo_point", GeoType::new);
@Getter
private String text;
@Getter
private String value;
private Supplier<DataType> typeBuilder;
public DataType getType() {
return typeBuilder.get();
}
public static ElasticPropertyType of(Object value) {
if (!StringUtils.isEmpty(value)) {
for (ElasticPropertyType elasticPropertyType : ElasticPropertyType.values()) {
if (elasticPropertyType.getValue().equals(value)) {
return elasticPropertyType;
}
}
}
return null;
}
public static ElasticPropertyType ofJava(Object value) {
if (!StringUtils.isEmpty(value)) {
for (ElasticPropertyType elasticPropertyType : ElasticPropertyType.values()) {
if (elasticPropertyType.getText().equals(value)) {
return elasticPropertyType;
}
}
}
throw new NotFoundException("未找到数据类型为:" + value + "的枚举");
}
}

View File

@ -1,57 +0,0 @@
package org.jetlinks.community.elastic.search.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.hswebframework.web.dict.EnumDict;
import org.hswebframework.web.exception.NotFoundException;
import org.springframework.util.StringUtils;
@AllArgsConstructor
@Getter
public enum FieldType implements EnumDict<String> {
TEXT("text", "text"),
BYTE("byte", "byte"),
SHORT("short", "short"),
INTEGER("int", "integer"),
LONG("long", "long"),
DATE("date", "date"),
HALF_FLOAT("half_float", "half_float"),
FLOAT("float", "float"),
DOUBLE("double", "double"),
BOOLEAN("boolean", "boolean"),
OBJECT("object", "object"),
AUTO("auto", "auto"),
NESTED("nested", "nested"),
IP("ip", "ip"),
ATTACHMENT("attachment", "attachment"),
KEYWORD("string", "keyword");
@Getter
private String text;
@Getter
private String value;
public static FieldType of(Object value) {
if (!StringUtils.isEmpty(value)) {
for (FieldType fieldType : FieldType.values()) {
if (fieldType.getValue().equals(value)) {
return fieldType;
}
}
}
throw new NotFoundException("未找到数据类型为:" + value + "的枚举");
}
public static FieldType ofJava(Object value) {
if (!StringUtils.isEmpty(value)) {
for (FieldType fieldType : FieldType.values()) {
if (fieldType.getText().equals(value)) {
return fieldType;
}
}
}
throw new NotFoundException("未找到数据类型为:" + value + "的枚举");
}
}

View File

@ -1,18 +0,0 @@
package org.jetlinks.community.elastic.search.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* @author bsetfeng
* @since 1.0
**/
@AllArgsConstructor
@Getter
public enum IndexPatternEnum {
MONTH("month"),
DAY("day");
private String value;
}

View File

@ -1,17 +0,0 @@
package org.jetlinks.community.elastic.search.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* @author bsetfeng
* @since 1.0
**/
@AllArgsConstructor
@Getter
public enum IndexStrategyEnum {
SUFFIX("suffix");
private String value;
}

View File

@ -1,68 +0,0 @@
package org.jetlinks.community.elastic.search.index;
import lombok.Getter;
import lombok.Setter;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.jetlinks.community.elastic.search.index.mapping.MappingFactory;
import org.jetlinks.community.elastic.search.index.setting.SettingFactory;
import java.util.Collections;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
public class CreateIndex {
@Getter
@Setter
private Map<String, Object> mapping;
@Getter
@Setter
private Settings.Builder settings;
private String index;
@Deprecated
private String type;
public CreateIndex addIndex(String index) {
this.index = index;
return this;
}
@Deprecated
public CreateIndex addType(String type) {
this.type = type;
return this;
}
public MappingFactory createMapping() {
return MappingFactory.createInstance(this);
}
public SettingFactory createSettings() {
return SettingFactory.createInstance(this);
}
public CreateIndexRequest createIndexRequest() {
CreateIndexRequest request = new CreateIndexRequest(index);
request.mapping(Collections.singletonMap("properties", getMapping()));
if (settings != null) {
request.settings(settings);
}
return request;
}
private CreateIndex() {
}
public static CreateIndex createInstance() {
return new CreateIndex();
}
}

View File

@ -0,0 +1,62 @@
package org.jetlinks.community.elastic.search.index;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
@ConfigurationProperties(prefix = "elasticsearch.index")
public class DefaultElasticSearchIndexManager implements ElasticSearchIndexManager {
@Getter
@Setter
private String defaultStrategy = "direct";
@Getter
@Setter
private Map<String, String> indexUseStrategy = new ConcurrentHashMap<>();
private Map<String, ElasticSearchIndexStrategy> strategies = new ConcurrentHashMap<>();
private Map<String, ElasticSearchIndexMetadata> indexMetadataStore = new ConcurrentHashMap<>();
public DefaultElasticSearchIndexManager(List<ElasticSearchIndexStrategy> strategies) {
strategies.forEach(this::registerStrategy);
}
@Override
public Mono<Void> putIndex(ElasticSearchIndexMetadata index) {
return this.getIndexStrategy(index.getIndex())
.flatMap(strategy -> strategy.putIndex(index))
.doOnSuccess(metadata -> indexMetadataStore.put(index.getIndex(), index));
}
@Override
public Mono<ElasticSearchIndexMetadata> getIndexMetadata(String index) {
return Mono.justOrEmpty(indexMetadataStore.get(index))
.switchIfEmpty(Mono.defer(() -> doLoadMetaData(index)
.doOnNext(metadata -> indexMetadataStore.put(metadata.getIndex(), metadata))));
}
protected Mono<ElasticSearchIndexMetadata> doLoadMetaData(String index) {
return getIndexStrategy(index)
.flatMap(strategy -> strategy.loadIndexMetadata(index));
}
@Override
public Mono<ElasticSearchIndexStrategy> getIndexStrategy(String index) {
return Mono.justOrEmpty(strategies.get(indexUseStrategy.getOrDefault(index.toLowerCase(), defaultStrategy)))
.switchIfEmpty(Mono.error(() -> new IllegalArgumentException("[" + index + "] 不支持任何索引策略")));
}
protected void registerStrategy(ElasticSearchIndexStrategy strategy) {
strategies.put(strategy.getId(), strategy);
}
}

View File

@ -0,0 +1,53 @@
package org.jetlinks.community.elastic.search.index;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.SimplePropertyMetadata;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class DefaultElasticSearchIndexMetadata implements ElasticSearchIndexMetadata {
private String index;
private Map<String, PropertyMetadata> properties = new HashMap<>();
public DefaultElasticSearchIndexMetadata(String index) {
this.index = index.toLowerCase().trim();
}
public DefaultElasticSearchIndexMetadata(String index, List<PropertyMetadata> properties) {
this(index);
properties.forEach(this::addProperty);
}
@Override
public PropertyMetadata getProperty(String property) {
return properties.get(property);
}
@Override
public String getIndex() {
return index;
}
@Override
public List<PropertyMetadata> getProperties() {
return new ArrayList<>(properties.values());
}
public DefaultElasticSearchIndexMetadata addProperty(PropertyMetadata property) {
properties.put(property.getId(), property);
return this;
}
public DefaultElasticSearchIndexMetadata addProperty(String property, DataType type) {
SimplePropertyMetadata metadata=new SimplePropertyMetadata();
metadata.setValueType(type);
metadata.setId(property);
properties.put(property, metadata);
return this;
}
}

View File

@ -1,136 +0,0 @@
package org.jetlinks.community.elastic.search.index;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.*;
import org.hswebframework.web.bean.FastBeanCopier;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
import org.jetlinks.community.elastic.search.index.mapping.IndicesMappingCenter;
import org.jetlinks.community.elastic.search.index.mapping.SingleMappingMetadata;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Mono;
import java.util.HashMap;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
@Service
@Slf4j
public class DefaultIndexOperationService implements IndexOperationService {
private final ElasticRestClient restClient;
private final IndicesMappingCenter indicesMappingCenter;
@Autowired
public DefaultIndexOperationService(ElasticRestClient restClient, IndicesMappingCenter indicesMappingCenter) {
this.restClient = restClient;
this.indicesMappingCenter = indicesMappingCenter;
}
@Override
public Mono<Boolean> indexIsExists(String index) {
return Mono.create(sink -> {
restClient.getQueryClient().indices().existsAsync(new GetIndexRequest(index), RequestOptions.DEFAULT, new ActionListener<Boolean>() {
@Override
public void onResponse(Boolean aBoolean) {
sink.success(aBoolean);
}
@Override
public void onFailure(Exception e) {
log.error("查询es index 是否存在失败", e);
sink.error(e);
}
});
});
}
@Override
public Mono<Boolean> init(CreateIndexRequest request) {
return indexIsExists(request.index())
.filter(bool -> !bool)
.flatMap(b -> Mono.create(sink -> {
restClient.getQueryClient().indices().createAsync(request, RequestOptions.DEFAULT, new ActionListener<CreateIndexResponse>() {
@Override
public void onResponse(CreateIndexResponse createIndexResponse) {
sink.success(createIndexResponse.isAcknowledged());
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
}));
}
@Override
public Mono<IndexMappingMetadata> getIndexMappingMetadata(String index) {
return indicesMappingCenter.getIndexMappingMetaData(index)
.map(Mono::just).orElseGet(() ->
getIndexMapping(index)
.doOnNext(indicesMappingCenter::register));
}
private Mono<IndexMappingMetadata> getIndexMapping(String index) {
return indexIsExists(index)
.filter(Boolean::booleanValue)
.flatMap(bool -> Mono.<IndexMappingMetadata>create(sink -> {
if (bool) {
GetMappingsRequest mappingsRequest = new GetMappingsRequest();
mappingsRequest.indices(index);
restClient.getQueryClient().indices().getMappingAsync(mappingsRequest, RequestOptions.DEFAULT, new ActionListener<GetMappingsResponse>() {
@Override
public void onResponse(GetMappingsResponse getMappingsResponse) {
//index存在时 getMappingsResponse.mappings().get(index).getSourceAsMap().get("properties") 不会为空
//sink.success(fieldMappingConvert(null, IndexMappingMetadata.getInstance(index), getMappingsResponse.mappings().get(index).getSourceAsMap().get("properties")));
getMappingsResponse.mappings()
.forEach((k, v) -> sink.success(fieldMappingConvert(null, IndexMappingMetadata.getInstance(index), v.getSourceAsMap().get("properties"))));
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
}
}))
.switchIfEmpty(Mono.just(IndexMappingMetadata.getInstance(index)));
}
private IndexMappingMetadata fieldMappingConvert(String baseKey, IndexMappingMetadata indexMappingMetaData, Object properties) {
FastBeanCopier.copy(properties, new HashMap<String, Object>())
.forEach((key, value) -> {
if (StringUtils.hasText(baseKey)) {
key = baseKey.concat(".").concat(key);
}
if (value instanceof Map) {
Map tempValue = FastBeanCopier.copy(value, new HashMap<>());
Object childProperties = tempValue.get("properties");
if (childProperties != null) {
fieldMappingConvert(key, indexMappingMetaData, childProperties);
return;
}
indexMappingMetaData.setMetadata(SingleMappingMetadata.builder()
.name(key)
.type(FieldType.of(tempValue.get("type")))
.build());
}
});
return indexMappingMetaData;
}
}

View File

@ -1,37 +1,9 @@
package org.jetlinks.community.elastic.search.index;
import java.util.function.Supplier;
/**
* @version 1.0
**/
public interface ElasticIndex {
@Deprecated
String getIndex();
@Deprecated
String getType();
default String getStandardIndex(){
return getIndex().toLowerCase();
}
default String getStandardType(){
return getType().toLowerCase();
}
static ElasticIndex createDefaultIndex(Supplier<String> indexConsumer, Supplier<String> typeConsumer) {
return new ElasticIndex() {
@Override
public String getIndex() {
return indexConsumer.get();
}
@Override
public String getType() {
return typeConsumer.get();
}
};
}
}

View File

@ -0,0 +1,13 @@
package org.jetlinks.community.elastic.search.index;
import reactor.core.publisher.Mono;
public interface ElasticSearchIndexManager {
Mono<Void> putIndex(ElasticSearchIndexMetadata index);
Mono<ElasticSearchIndexMetadata> getIndexMetadata(String index);
Mono<ElasticSearchIndexStrategy> getIndexStrategy(String index);
}

View File

@ -0,0 +1,28 @@
package org.jetlinks.community.elastic.search.index;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
import java.util.List;
import java.util.Map;
public interface ElasticSearchIndexMetadata {
String getIndex();
List<PropertyMetadata> getProperties();
PropertyMetadata getProperty(String property);
default Map<String, Object> convertToElastic(Map<String, Object> map) {
return ElasticSearchConverter.convertDataToElastic(map, getProperties());
}
default Map<String, Object> convertFromElastic(Map<String, Object> map) {
return ElasticSearchConverter.convertDataFromElastic(map, getProperties());
}
default ElasticSearchIndexMetadata newIndexName(String name) {
return new DefaultElasticSearchIndexMetadata(name, getProperties());
}
}

View File

@ -0,0 +1,45 @@
package org.jetlinks.community.elastic.search.index;
import reactor.core.publisher.Mono;
/**
* es 索引策略
*
* @author zhouhao
* @version 1.0
*/
public interface ElasticSearchIndexStrategy {
/**
* 策略标识
*
* @return ID
*/
String getId();
/**
* 获取用于获取保存数据的索引
*
* @param index 原始索引名
* @return 索引名
*/
String getIndexForSave(String index);
/**
* 获取用于搜索的索引
*
* @param index 原始索引名
* @return 索引名
*/
String getIndexForSearch(String index);
/**
* 更新索引
*
* @param metadata 索引元数据
* @return 更新结果
*/
Mono<Void> putIndex(ElasticSearchIndexMetadata metadata);
Mono<ElasticSearchIndexMetadata> loadIndexMetadata(String index);
}

View File

@ -1,66 +0,0 @@
package org.jetlinks.community.elastic.search.index.alias;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.service.IndexAliasOperationService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
/**
* @author bsetfeng
* @since 1.0
**/
@Service
@Slf4j
public class DefaultIndexAliasOperationService implements IndexAliasOperationService {
private final ElasticRestClient restClient;
@Autowired
public DefaultIndexAliasOperationService(ElasticRestClient restClient) {
this.restClient = restClient;
}
@Override
public Mono<Boolean> indexAliasIsExists(String alias) {
return Mono.create(sink -> {
GetAliasesRequest request = new GetAliasesRequest(alias);
restClient.getQueryClient().indices().existsAliasAsync(request, RequestOptions.DEFAULT, new ActionListener<Boolean>() {
@Override
public void onResponse(Boolean aBoolean) {
sink.success(aBoolean);
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
});
}
@Override
public Mono<Boolean> AddAlias(IndicesAliasesRequest request) {
return Mono.create(sink -> {
restClient.getQueryClient().indices().updateAliasesAsync(request, RequestOptions.DEFAULT, new ActionListener<AcknowledgedResponse>() {
@Override
public void onResponse(AcknowledgedResponse acknowledgedResponse) {
sink.success(acknowledgedResponse.isAcknowledged());
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
});
}
}

View File

@ -2,7 +2,7 @@ package org.jetlinks.community.elastic.search.index.mapping;
import lombok.Getter;
import lombok.Setter;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.enums.ElasticPropertyType;
import java.util.HashMap;
import java.util.List;
@ -33,14 +33,14 @@ public class IndexMappingMetadata {
return metadata.get(fieldName);
}
public List<SingleMappingMetadata> getMetaDataByType(FieldType type) {
public List<SingleMappingMetadata> getMetaDataByType(ElasticPropertyType type) {
return getAllMetaData()
.stream()
.filter(singleMapping -> singleMapping.getType().equals(type))
.collect(Collectors.toList());
}
public Map<String, SingleMappingMetadata> getMetaDataByTypeToMap(FieldType type) {
public Map<String, SingleMappingMetadata> getMetaDataByTypeToMap(ElasticPropertyType type) {
return getMetaDataByType(type)
.stream()
.collect(Collectors.toMap(SingleMappingMetadata::getName, Function.identity()));

View File

@ -1,26 +0,0 @@
package org.jetlinks.community.elastic.search.index.mapping;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class IndicesMappingCenter {
private Map<String, IndexMappingMetadata> indicesMapping = new ConcurrentHashMap<>();
public Optional<IndexMappingMetadata> getIndexMappingMetaData(String index) {
return Optional.ofNullable(indicesMapping.get(index));
}
public void register(IndexMappingMetadata mappingMetaData) {
indicesMapping.put(mappingMetaData.getIndex(), mappingMetaData);
}
}

View File

@ -1,77 +0,0 @@
package org.jetlinks.community.elastic.search.index.mapping;
import org.hswebframework.web.exception.BusinessException;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
public class MappingFactory {
private Map<String, Object> properties = new HashMap<>();
private Map<String, Object> filedMap = new HashMap<>();
private volatile boolean flag = true;
private CreateIndex index;
private MappingFactory(CreateIndex index) {
this.index = index;
}
public MappingFactory addFieldName(String fieldName) {
continuityOperateHandle(!flag);
flag = false;
filedMap = new HashMap<>();
properties.put(fieldName, filedMap);
return this;
}
public MappingFactory addFieldType(FieldType type) {
continuityOperateHandle(flag);
filedMap.put("type", type.getValue());
return this;
}
public MappingFactory addFieldDateFormat(FieldDateFormat... dateFormats) {
continuityOperateHandle(flag);
filedMap.compute("format", (k, v) -> v == null ? FieldDateFormat.getFormat(Arrays.asList(dateFormats)) : v + "||" + FieldDateFormat.getFormat(Arrays.asList(dateFormats)));
return this;
}
public MappingFactory addFieldDateFormat(String... dateFormats) {
continuityOperateHandle(flag);
filedMap.compute("format", (k, v) -> v == null ? FieldDateFormat.getFormatStr(Arrays.asList(dateFormats)) : v + "||" + FieldDateFormat.getFormatStr(Arrays.asList(dateFormats)));
return this;
}
public MappingFactory commit() {
flag = true;
return this;
}
public CreateIndex end() {
index.setMapping(properties);
return index;
}
public static MappingFactory createInstance(CreateIndex index) {
return new MappingFactory(index);
}
private void continuityOperateHandle(boolean inoperable) {
if (inoperable) {
throw new BusinessException("please exec commit() or addFiledName() later then operate");
}
}
}

View File

@ -1,8 +1,8 @@
package org.jetlinks.community.elastic.search.index.mapping;
import lombok.*;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.enums.ElasticDateFormat;
import org.jetlinks.community.elastic.search.enums.ElasticPropertyType;
/**
* @author bsetfeng
@ -17,7 +17,7 @@ public class SingleMappingMetadata {
private String name;
private FieldDateFormat format;
private ElasticDateFormat format;
private FieldType type;
private ElasticPropertyType type;
}

View File

@ -1,40 +0,0 @@
package org.jetlinks.community.elastic.search.index.setting;
import org.elasticsearch.common.settings.Settings;
import org.jetlinks.community.elastic.search.index.CreateIndex;
/**
* @author bsetfeng
* @since 1.0
**/
public class SettingFactory {
private Settings.Builder settings = Settings.builder();
private CreateIndex index;
private SettingFactory(CreateIndex index) {
this.index = index;
}
public SettingFactory settingShards(Integer shards) {
settings.put("number_of_shards", shards);
return this;
}
public SettingFactory settingReplicas(Integer replicas) {
settings.put("number_of_replicas", replicas);
return this;
}
public CreateIndex end() {
index.setSettings(settings);
return index;
}
public static SettingFactory createInstance(CreateIndex index) {
return new SettingFactory(index);
}
}

View File

@ -0,0 +1,187 @@
package org.jetlinks.community.elastic.search.index.strategies;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.*;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.SimplePropertyMetadata;
import org.jetlinks.core.metadata.types.*;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.enums.ElasticDateFormat;
import org.jetlinks.community.elastic.search.enums.ElasticPropertyType;
import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexStrategy;
import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
import reactor.core.publisher.Mono;
import java.util.*;
import java.util.stream.Collectors;
@AllArgsConstructor
@Slf4j
public abstract class AbstractElasticSearchIndexStrategy implements ElasticSearchIndexStrategy {
@Getter
private String id;
protected ElasticRestClient client;
private String wrapIndex(String index) {
return index.toLowerCase();
}
public Mono<Boolean> indexExists(String index) {
return ReactorActionListener.mono(
actionListener ->
client.getQueryClient()
.indices()
.existsAsync(new GetIndexRequest(wrapIndex(index)), RequestOptions.DEFAULT, actionListener));
}
public Mono<Void> doCreateIndex(ElasticSearchIndexMetadata metadata) {
return ReactorActionListener.<CreateIndexResponse>mono(
actionListener -> client.getQueryClient()
.indices()
.createAsync(createIndexRequest(metadata), RequestOptions.DEFAULT, actionListener))
.then();
}
protected Mono<Void> doPutIndex(ElasticSearchIndexMetadata metadata,
boolean justUpdateMapping) {
String index = wrapIndex(metadata.getIndex());
return this.indexExists(index)
.flatMap(exists -> {
if (exists) {
return doLoadIndexMetadata(index)
.flatMap(oldMapping -> Mono.justOrEmpty(createPutMappingRequest(metadata, oldMapping)))
.flatMap(request -> ReactorActionListener.<AcknowledgedResponse>mono(
actionListener ->
client.getWriteClient()
.indices()
.putMappingAsync(request, RequestOptions.DEFAULT, actionListener)))
.then();
}
if (justUpdateMapping) {
return Mono.empty();
}
return doCreateIndex(metadata);
});
}
protected Mono<ElasticSearchIndexMetadata> doLoadIndexMetadata(String _index) {
String index = wrapIndex(_index);
return ReactorActionListener
.<GetMappingsResponse>mono(listener -> client.getQueryClient()
.indices()
.getMappingAsync(new GetMappingsRequest().indices(index), RequestOptions.DEFAULT, listener))
.flatMap(resp -> Mono.justOrEmpty(convertMetadata(index, resp.mappings().get(index))));
}
public CreateIndexRequest createIndexRequest(ElasticSearchIndexMetadata metadata) {
CreateIndexRequest request = new CreateIndexRequest(wrapIndex(metadata.getIndex()));
request.mapping(Collections.singletonMap("properties", createElasticProperties(metadata.getProperties())));
return request;
}
private PutMappingRequest createPutMappingRequest(ElasticSearchIndexMetadata metadata, ElasticSearchIndexMetadata ignore) {
Map<String, Object> properties = createElasticProperties(metadata.getProperties());
Map<String, Object> ignoreProperties = createElasticProperties(ignore.getProperties());
for (Map.Entry<String, Object> en : ignoreProperties.entrySet()) {
log.debug("ignore update index [{}] mapping property:{},{}", wrapIndex(metadata.getIndex()), en.getKey(), en.getValue());
properties.remove(en.getKey());
}
if (properties.isEmpty()) {
log.debug("ignore update index [{}] mapping", wrapIndex(metadata.getIndex()));
return null;
}
PutMappingRequest request = new PutMappingRequest(wrapIndex(metadata.getIndex()));
request.source(Collections.singletonMap("properties", properties));
return request;
}
protected Map<String, Object> createElasticProperties(List<PropertyMetadata> metadata) {
if (metadata == null) {
return new HashMap<>();
}
return metadata.stream()
.collect(Collectors.toMap(PropertyMetadata::getId, this::createElasticProperty));
}
protected Map<String, Object> createElasticProperty(PropertyMetadata metadata) {
Map<String, Object> property = new HashMap<>();
DataType type = metadata.getValueType();
if (type instanceof DateTimeType) {
property.put("type", "date");
property.put("format", ElasticDateFormat.getFormat(ElasticDateFormat.epoch_millis, ElasticDateFormat.simple_date, ElasticDateFormat.strict_date));
} else if (type instanceof DoubleType) {
property.put("type", "double");
} else if (type instanceof LongType) {
property.put("type", "long");
} else if (type instanceof IntType) {
property.put("type", "integer");
} else if (type instanceof FloatType) {
property.put("type", "float");
} else if (type instanceof BooleanType) {
property.put("type", "boolean");
} else if (type instanceof GeoType) {
property.put("type", "geo_point");
} else if (type instanceof ObjectType) {
property.put("type", "nested");
ObjectType objectType = ((ObjectType) type);
property.put("properties", createElasticProperties(objectType.getProperties()));
} else {
property.put("type", "keyword");
}
return property;
}
protected ElasticSearchIndexMetadata convertMetadata(String index, MappingMetaData metaData) {
Map<String, Object> response = metaData.getSourceAsMap();
Object properties = response.get("properties");
return new DefaultElasticSearchIndexMetadata(index, convertProperties(properties));
}
@SuppressWarnings("all")
protected List<PropertyMetadata> convertProperties(Object properties) {
if (properties == null) {
return new ArrayList<>();
}
return ((Map<String, Map<String, Object>>) properties)
.entrySet()
.stream()
.map(entry -> convertProperty(entry.getKey(), entry.getValue()))
.collect(Collectors.toList());
}
private PropertyMetadata convertProperty(String property, Map<String, Object> map) {
String type = String.valueOf(map.get("type"));
SimplePropertyMetadata metadata = new SimplePropertyMetadata();
metadata.setId(property);
metadata.setName(property);
ElasticPropertyType elasticPropertyType = ElasticPropertyType.of(type);
if (null != elasticPropertyType) {
DataType dataType = elasticPropertyType.getType();
if ((elasticPropertyType == ElasticPropertyType.OBJECT
|| elasticPropertyType == ElasticPropertyType.NESTED)
&& dataType instanceof ObjectType) {
@SuppressWarnings("all")
Map<String, Object> nestProperties = (Map<String, Object>) map.get("properties");
if (null != nestProperties) {
ObjectType objectType = ((ObjectType) dataType);
objectType.setProperties(convertProperties(nestProperties));
}
}
metadata.setValueType(dataType);
} else {
metadata.setValueType(new StringType());
}
return metadata;
}
}

View File

@ -0,0 +1,35 @@
package org.jetlinks.community.elastic.search.index.strategies;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
@Component
public class DirectElasticSearchIndexStrategy extends AbstractElasticSearchIndexStrategy {
public DirectElasticSearchIndexStrategy(ElasticRestClient client) {
super("direct", client);
}
@Override
public String getIndexForSave(String index) {
return index;
}
@Override
public String getIndexForSearch(String index) {
return index;
}
@Override
public Mono<Void> putIndex(ElasticSearchIndexMetadata metadata) {
return doPutIndex(metadata, false);
}
@Override
public Mono<ElasticSearchIndexMetadata> loadIndexMetadata(String index) {
return doLoadIndexMetadata(index);
}
}

View File

@ -0,0 +1,71 @@
package org.jetlinks.community.elastic.search.index.strategies;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.GetIndexTemplatesRequest;
import org.elasticsearch.client.indices.GetIndexTemplatesResponse;
import org.elasticsearch.client.indices.PutIndexTemplateRequest;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
import reactor.core.publisher.Mono;
import java.util.Collections;
import java.util.List;
public abstract class TemplateElasticSearchIndexStrategy extends AbstractElasticSearchIndexStrategy {
public TemplateElasticSearchIndexStrategy(String id, ElasticRestClient client) {
super(id, client);
}
protected String getTemplate(String index) {
return index.concat("_template");
}
protected String getAlias(String index) {
return index.concat("_alias");
}
protected List<String> getIndexPatterns(String index) {
return Collections.singletonList(index.concat("*"));
}
@Override
public abstract String getIndexForSave(String index);
@Override
public String getIndexForSearch(String index) {
return getAlias(index);
}
@Override
public Mono<Void> putIndex(ElasticSearchIndexMetadata metadata) {
return ReactorActionListener
.<AcknowledgedResponse>mono(listener -> client.getWriteClient()
.indices()//修改索引模版
.putTemplateAsync(createIndexTemplateRequest(metadata), RequestOptions.DEFAULT, listener))
//修改当前索引
.then(doPutIndex(metadata.newIndexName(getIndexForSave(metadata.getIndex())), true));
}
protected PutIndexTemplateRequest createIndexTemplateRequest(ElasticSearchIndexMetadata metadata) {
String index = metadata.getIndex();
PutIndexTemplateRequest request = new PutIndexTemplateRequest(getTemplate(index));
request.alias(new Alias(getAlias(index)));
request.mapping(Collections.singletonMap("properties", createElasticProperties(metadata.getProperties())));
request.patterns(getIndexPatterns(index));
return request;
}
@Override
public Mono<ElasticSearchIndexMetadata> loadIndexMetadata(String index) {
return ReactorActionListener
.<GetIndexTemplatesResponse>mono(listener -> client.getQueryClient()
.indices()
.getIndexTemplateAsync(new GetIndexTemplatesRequest(getTemplate(index)), RequestOptions.DEFAULT, listener))
.filter(resp -> resp.getIndexTemplates().size() > 0)
.flatMap(resp -> Mono.justOrEmpty(convertMetadata(index, resp.getIndexTemplates().get(0).mappings())));
}
}

View File

@ -0,0 +1,28 @@
package org.jetlinks.community.elastic.search.index.strategies;
import org.hswebframework.utils.time.DateFormatter;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* 按月对来划分索引策略
*
* @author zhouhao
* @since 1.0
*/
@Component
public class TimeByMonthElasticSearchIndexStrategy extends TemplateElasticSearchIndexStrategy {
private final String format = "yyyy-MM";
public TimeByMonthElasticSearchIndexStrategy(ElasticRestClient client) {
super("time-by-month", client);
}
@Override
public String getIndexForSave(String index) {
return index.concat("_").concat(DateFormatter.toString(new Date(), format));
}
}

View File

@ -1,67 +0,0 @@
package org.jetlinks.community.elastic.search.index.template;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.IndexTemplatesExistRequest;
import org.elasticsearch.client.indices.PutIndexTemplateRequest;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.service.IndexTemplateOperationService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
/**
* @author bsetfeng
* @since 1.0
**/
@Service
@Slf4j
public class DefaultIndexTemplateOperationService implements IndexTemplateOperationService {
private final ElasticRestClient restClient;
@Autowired
public DefaultIndexTemplateOperationService(ElasticRestClient restClient) {
this.restClient = restClient;
}
@Override
public Mono<Boolean> indexTemplateIsExists(String name) {
return Mono.create(sink -> {
IndexTemplatesExistRequest request = new IndexTemplatesExistRequest(name);
restClient.getQueryClient().indices().existsTemplateAsync(request, RequestOptions.DEFAULT, new ActionListener<Boolean>() {
@Override
public void onResponse(Boolean aBoolean) {
sink.success(aBoolean);
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
});
}
@Override
public Mono<Boolean> putTemplate(PutIndexTemplateRequest request) {
return indexTemplateIsExists(request.name())
.filter(bool -> !bool)
.flatMap(b -> Mono.create(sink -> {
restClient.getQueryClient().indices().putTemplateAsync(request, RequestOptions.DEFAULT, new ActionListener<AcknowledgedResponse>() {
@Override
public void onResponse(AcknowledgedResponse acknowledgedResponse) {
sink.success(acknowledgedResponse.isAcknowledged());
}
@Override
public void onFailure(Exception e) {
sink.error(e);
}
});
}));
}
}

View File

@ -1,50 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
import lombok.Getter;
import org.jetlinks.community.elastic.search.enums.IndexPatternEnum;
import org.jetlinks.community.elastic.search.enums.IndexStrategyEnum;
import org.jetlinks.community.elastic.search.manager.entity.IndexStrategy;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
@Getter
public class DefaultIndexStrategyProvider implements IndexStrategyProvider {
@Value("${elasticsearch.time-series.index-strategy.suffix-month:}")
private List<String> suffixMonthIndices;
@Value("${elasticsearch.time-series.index-strategy.suffix-day:}")
private List<String> suffixDayIndices;
@Value("${elasticsearch.time-series.index-strategy.format:yyyy-MM}")
private String format;
@Value("${elasticsearch.time-series.index-strategy.connector:-}")
private String connector;
@Override
public Map<String, IndexStrategy> getIndexStrategies() {
Map<String, IndexStrategy> indexStrategyMap = new HashMap<>();
suffixMonthIndices
.stream()
.filter(StringUtils::hasText)
.forEach(s -> indexStrategyMap.put(s, new IndexStrategy(IndexStrategyEnum.SUFFIX.getValue(), IndexPatternEnum.MONTH.getValue())));
suffixMonthIndices
.stream()
.filter(StringUtils::hasText)
.forEach(s -> indexStrategyMap.put(s, new IndexStrategy(IndexStrategyEnum.SUFFIX.getValue(), IndexPatternEnum.DAY.getValue())));
return indexStrategyMap;
}
}

View File

@ -1,20 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexManager {
IndexPatternManager getIndexPatternManager();
IndexStrategyManager getIndexStrategyManager();
String getFormat();
String getConnector();
default String getStandardIndex(String index) {
return getIndexStrategyManager().getStandardsIndex(index, getIndexPatternManager().getPattern(getFormat()), getConnector());
}
}

View File

@ -1,12 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexPatternManager {
String getName();
String getPattern(String format);
}

View File

@ -1,12 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexStrategyManager {
String getName();
String getStandardsIndex(String index, String pattern, String connector);
}

View File

@ -1,37 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
import org.jetlinks.community.elastic.search.manager.entity.IndexStrategy;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexStrategyProvider {
/**
* index 时间格式 ps: yyyy-MM-dd
*
* @return
*/
String getFormat();
/**
* index与策略串的连接符
*
* @return
*/
default String connector() {
return "-";
}
/**
* @see StandardsIndexManagerCenter
*
* @return
*/
Map<String, IndexStrategy> getIndexStrategies();
}

View File

@ -1,21 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
import org.springframework.stereotype.Component;
import java.util.concurrent.ScheduledExecutorService;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class StandardsIndexInit {
private final ScheduledExecutorService executorService;
public StandardsIndexInit(ScheduledExecutorService executorService) {
this.executorService = executorService;
}
}

View File

@ -1,21 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
/**
* @author bsetfeng
* @since 1.0
**/
public interface StandardsIndexManager {
String getStandardsIndex(String index);
boolean indexIsChange(String index);
boolean indexIsUpdate(String index);
boolean standardsIndexIsUpdate(String standardsIndex);
void addStandardsIndex(String standardsIndex);
public void registerIndexManager(String index, IndexManager indexManager);
}

View File

@ -1,124 +0,0 @@
package org.jetlinks.community.elastic.search.manager;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class StandardsIndexManagerCenter implements StandardsIndexManager, BeanPostProcessor, CommandLineRunner {
Map<String, IndexManager> managerMap = new ConcurrentHashMap<>();
Map<String, IndexPatternManager> patternManagerMap = new ConcurrentHashMap<>();
Map<String, IndexStrategyManager> strategyManagerMap = new ConcurrentHashMap<>();
Set<String> standardsIndices = new ConcurrentSkipListSet<>();
private final List<IndexStrategyProvider> indexStrategyProviders;
public StandardsIndexManagerCenter(List<IndexStrategyProvider> indexStrategyProviders) {
this.indexStrategyProviders = indexStrategyProviders;
}
public void registerIndexPatternManager(IndexPatternManager patternManager) {
patternManagerMap.put(patternManager.getName(), patternManager);
}
public void registerIndexStrategyManager(IndexStrategyManager strategyManager) {
strategyManagerMap.put(strategyManager.getName(), strategyManager);
}
@Override
public void registerIndexManager(String index, IndexManager indexManager) {
managerMap.put(index, indexManager);
}
@Override
public String getStandardsIndex(String index) {
return Optional.ofNullable(managerMap.get(index))
.map(m -> {
String standardsIndex = m.getStandardIndex(index);
standardsIndices.add(standardsIndex);
return standardsIndex;
})
.orElse(index);
}
@Override
public boolean indexIsChange(String index) {
return managerMap.containsKey(index);
}
@Override
public boolean indexIsUpdate(String index) {
return Optional.ofNullable(managerMap.get(index))
.map(m -> !standardsIndices.contains(m.getStandardIndex(index)))
.orElse(false);
}
@Override
public boolean standardsIndexIsUpdate(String standardsIndex) {
return !standardsIndices.contains(standardsIndex);
}
@Override
public void addStandardsIndex(String standardsIndex) {
standardsIndices.add(standardsIndex);
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof IndexPatternManager) {
registerIndexPatternManager((IndexPatternManager) bean);
}
if (bean instanceof IndexStrategyManager) {
registerIndexStrategyManager((IndexStrategyManager) bean);
}
return bean;
}
@Override
public void run(String... args) throws Exception {
indexStrategyProviders.forEach(provider -> {
provider.getIndexStrategies()
.forEach((k, v) -> managerMap.put(k, new IndexManager() {
@Override
public IndexPatternManager getIndexPatternManager() {
return patternManagerMap.get(v.getPatternName());
}
@Override
public IndexStrategyManager getIndexStrategyManager() {
return strategyManagerMap.get(v.getStrategyName());
}
@Override
public String getFormat() {
return provider.getFormat();
}
@Override
public String getConnector() {
return provider.connector();
}
}));
});
}
}

View File

@ -1,19 +0,0 @@
package org.jetlinks.community.elastic.search.manager.entity;
import lombok.*;
/**
* @author bsetfeng
* @since 1.0
**/
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class IndexStrategy {
private String strategyName;
private String patternName;
}

View File

@ -1,26 +0,0 @@
package org.jetlinks.community.elastic.search.manager.strategy;
import org.jetlinks.community.elastic.search.enums.IndexPatternEnum;
import org.jetlinks.community.elastic.search.manager.IndexPatternManager;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class IndexDayPattern implements IndexPatternManager {
@Override
public String getName() {
return IndexPatternEnum.DAY.getValue();
}
@Override
public String getPattern(String format) {
LocalDateTime localDateTime = LocalDateTime.now();
return localDateTime.format(DateTimeFormatter.ofPattern(format));
}
}

View File

@ -1,27 +0,0 @@
package org.jetlinks.community.elastic.search.manager.strategy;
import org.jetlinks.community.elastic.search.enums.IndexPatternEnum;
import org.jetlinks.community.elastic.search.manager.IndexPatternManager;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class IndexMonthPattern implements IndexPatternManager {
@Override
public String getName() {
return IndexPatternEnum.MONTH.getValue();
}
@Override
public String getPattern(String format) {
LocalDateTime localDateTime = LocalDateTime.now();
return localDateTime.format(DateTimeFormatter.ofPattern(format));
}
}

View File

@ -1,24 +0,0 @@
package org.jetlinks.community.elastic.search.manager.strategy;
import org.jetlinks.community.elastic.search.enums.IndexStrategyEnum;
import org.jetlinks.community.elastic.search.manager.IndexStrategyManager;
import org.springframework.stereotype.Component;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class IndexSuffixStrategy implements IndexStrategyManager {
@Override
public String getName() {
return IndexStrategyEnum.SUFFIX.getValue();
}
@Override
public String getStandardsIndex(String index, String pattern, String connector) {
return index.concat(connector).concat(pattern);
}
}

View File

@ -1,35 +0,0 @@
package org.jetlinks.community.elastic.search.parser;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
import org.springframework.util.StringUtils;
/**
* @author bsetfeng
* @since 1.0
**/
public abstract class AbstractQueryParamTranslateService implements QueryParamTranslateService {
@Override
public SearchSourceBuilder translate(QueryParam queryParam, IndexMappingMetadata mappingMetaData) {
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
if (queryParam.isPaging()) {
sourceBuilder.from(queryParam.getPageIndex() * queryParam.getPageSize());
sourceBuilder.size(queryParam.getPageSize());
}
queryParam.getSorts()
.forEach(sort -> {
if (!StringUtils.isEmpty(sort.getName())) {
sourceBuilder.sort(sort.getName(), SortOrder.fromString(sort.getOrder()));
}
});
return sourceBuilder.query(queryBuilder(queryParam, mappingMetaData));
}
protected abstract QueryBuilder queryBuilder(QueryParam queryParam, IndexMappingMetadata mappingMetaData);
}

View File

@ -5,7 +5,6 @@ import org.elasticsearch.index.query.QueryBuilders;
import org.hswebframework.ezorm.core.param.Term;
import org.springframework.stereotype.Component;
import java.util.LinkedList;
import java.util.List;
import java.util.function.Consumer;
@ -20,13 +19,10 @@ public class DefaultLinkTypeParser implements LinkTypeParser {
@Override
public BoolQueryBuilder process(Term term, Consumer<Term> consumer, BoolQueryBuilder queryBuilders) {
if ("or".equalsIgnoreCase(term.getType().name())) {
if (term.getType() == Term.Type.or) {
handleOr(queryBuilders, term, consumer);
} else if ("and".equalsIgnoreCase(term.getType().name())) {
handleAnd(queryBuilders, term, consumer);
} else {
throw new UnsupportedOperationException("不支持的查询连接类型,term.getType:" + term.getType().name());
handleAnd(queryBuilders, term, consumer);
}
return queryBuilders;
}
@ -37,7 +33,7 @@ public class DefaultLinkTypeParser implements LinkTypeParser {
parser.process(() -> term, queryBuilders::should);
} else {
BoolQueryBuilder nextQuery = QueryBuilders.boolQuery();
List<Term> terms = ( term.getTerms());
List<Term> terms = (term.getTerms());
terms.forEach(t -> process(t, consumer, nextQuery));
queryBuilders.should(nextQuery);
}
@ -45,7 +41,7 @@ public class DefaultLinkTypeParser implements LinkTypeParser {
private void handleAnd(BoolQueryBuilder queryBuilders, Term term, Consumer<Term> consumer) {
consumer.accept(term);
if (term.getTerms().isEmpty()&& term.getValue() != null) {
if (term.getTerms().isEmpty() && term.getValue() != null) {
parser.process(() -> term, queryBuilders::must);
} else {
BoolQueryBuilder nextQuery = QueryBuilders.boolQuery();
@ -54,4 +50,6 @@ public class DefaultLinkTypeParser implements LinkTypeParser {
queryBuilders.must(nextQuery);
}
}
}

View File

@ -1,52 +0,0 @@
package org.jetlinks.community.elastic.search.parser;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.core.param.Term;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
import org.jetlinks.community.elastic.search.index.mapping.SingleMappingMetadata;
import org.jetlinks.community.elastic.search.utils.DateTimeUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class DefaultQueryParamTranslateService extends AbstractQueryParamTranslateService {
private final LinkTypeParser parser;
@Value("${jetlinks.system.formats:yyyy-MM-dd HH:mm:ss}")
private List<String> formats;
@Autowired
public DefaultQueryParamTranslateService(LinkTypeParser parser) {
this.parser = parser;
}
@Override
protected QueryBuilder queryBuilder(QueryParam queryParam, IndexMappingMetadata mappingMetaData) {
BoolQueryBuilder queryBuilders = QueryBuilders.boolQuery();
queryParam.getTerms()
.forEach(term -> parser.process(term, t ->
dateTypeHandle(t, mappingMetaData.getMetaData(t.getColumn())), queryBuilders));
return queryBuilders;
}
private void dateTypeHandle(Term term, SingleMappingMetadata singleMappingMetaData) {
if (singleMappingMetaData == null) return;
if (singleMappingMetaData.getType().equals(FieldType.DATE)) {
term.setValue(DateTimeUtils.formatDateToTimestamp(term.getValue(), formats));
}
}
}

View File

@ -1,14 +0,0 @@
package org.jetlinks.community.elastic.search.parser;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
/**
* @author bsetfeng
* @since 1.0
**/
public interface QueryParamTranslateService {
SearchSourceBuilder translate(QueryParam queryParam, IndexMappingMetadata metaData);
}

View File

@ -6,15 +6,25 @@ import org.jetlinks.community.elastic.search.aggreation.bucket.BucketResponse;
import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsAggregationStructure;
import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsResponse;
import org.jetlinks.community.elastic.search.index.ElasticIndex;
import org.jetlinks.community.timeseries.query.AggregationQueryParam;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
public interface AggregationService {
Mono<MetricsResponse> metricsAggregation(QueryParam queryParam, MetricsAggregationStructure structure, ElasticIndex provider);
Mono<MetricsResponse> metricsAggregation(String index, QueryParam queryParam, MetricsAggregationStructure structure);
Mono<BucketResponse> bucketAggregation(QueryParam queryParam, BucketAggregationsStructure structure, ElasticIndex provider);
Mono<BucketResponse> bucketAggregation(String index, QueryParam queryParam, BucketAggregationsStructure structure);
Flux<Map<String, Object>> aggregation(String index, AggregationQueryParam queryParam);
default Flux<Map<String, Object>> aggregation(ElasticIndex index, AggregationQueryParam queryParam) {
return aggregation(index.getIndex(), queryParam);
}
}

View File

@ -1,13 +1,9 @@
package org.jetlinks.community.elastic.search.service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
@ -19,26 +15,31 @@ import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.utils.time.DateFormatter;
import org.hswebframework.utils.time.DefaultDateFormatter;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.jetlinks.core.utils.FluxUtils;
import org.jetlinks.community.elastic.search.ElasticRestClient;
import org.jetlinks.community.elastic.search.index.ElasticIndex;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
import org.jetlinks.community.elastic.search.parser.QueryParamTranslateService;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
import org.reactivestreams.Publisher;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuples;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.*;
import java.util.function.Function;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
* @author bsetfeng
* @author zhouhao
* @since 1.0
**/
@Service
@ -47,26 +48,35 @@ public class DefaultElasticSearchService implements ElasticSearchService {
private final ElasticRestClient restClient;
private final IndexOperationService indexOperationService;
private final QueryParamTranslateService translateService;
private final ElasticSearchIndexManager indexManager;
FluxSink<Buffer> sink;
static {
DateFormatter.supportFormatter.add(new DefaultDateFormatter(Pattern.compile("[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.+"), "yyyy-MM-dd'T'HH:mm:ss.SSSZ"));
}
public DefaultElasticSearchService(ElasticRestClient restClient,
QueryParamTranslateService translateService,
IndexOperationService indexOperationService) {
ElasticSearchIndexManager indexManager) {
this.restClient = restClient;
this.translateService = translateService;
this.indexOperationService = indexOperationService;
init();
this.indexManager = indexManager;
}
public <T> Flux<T> query(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
return doSearch(createSearchRequest(queryParam, index))
.flatMapIterable(response -> translate(mapper, response))
.onErrorResume(err -> {
log.error("query elastic error", err);
return Mono.empty();
});
}
@Override
public <T> Mono<PagerResult<T>> queryPager(ElasticIndex index, QueryParam queryParam, Class<T> type) {
return query(searchRequestStructure(queryParam, index))
.map(response -> translatePageResult(type, queryParam, response))
public <T> Mono<PagerResult<T>> queryPager(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
return doSearch(createSearchRequest(queryParam, index))
.map(response -> translatePageResult(mapper, queryParam, response))
.switchIfEmpty(Mono.just(PagerResult.empty()))
.onErrorReturn(err -> {
log.error("query elastic error", err);
@ -75,18 +85,10 @@ public class DefaultElasticSearchService implements ElasticSearchService {
}
@Override
public <T> Flux<T> query(ElasticIndex index, QueryParam queryParam, Class<T> type) {
return query(searchRequestStructure(queryParam, index))
.flatMapIterable(response -> translate(type, response))
.onErrorResume(err -> {
log.error("query elastic error", err);
return Flux.empty();
});
}
@Override
public Mono<Long> count(ElasticIndex index, QueryParam queryParam) {
return countQuery(countRequestStructure(queryParam, index))
public Mono<Long> count(String index, QueryParam queryParam) {
QueryParam param=queryParam.clone();
param.setPaging(false);
return doCount(createCountRequest(param, index))
.map(CountResponse::getCount)
.defaultIfEmpty(0L)
.onErrorReturn(err -> {
@ -95,16 +97,15 @@ public class DefaultElasticSearchService implements ElasticSearchService {
}, 0L);
}
@Override
public <T> Mono<Void> commit(ElasticIndex index, T payload) {
public <T> Mono<Void> commit(String index, T payload) {
return Mono.fromRunnable(() -> {
sink.next(new Buffer(index, payload));
});
}
@Override
public <T> Mono<Void> commit(ElasticIndex index, Collection<T> payload) {
public <T> Mono<Void> commit(String index, Collection<T> payload) {
return Mono.fromRunnable(() -> {
for (T t : payload) {
sink.next(new Buffer(index, t));
@ -113,7 +114,7 @@ public class DefaultElasticSearchService implements ElasticSearchService {
}
@Override
public <T> Mono<Void> commit(ElasticIndex index, Publisher<T> data) {
public <T> Mono<Void> commit(String index, Publisher<T> data) {
return Flux.from(data)
.flatMap(d -> commit(index, d))
.then();
@ -148,34 +149,51 @@ public class DefaultElasticSearchService implements ElasticSearchService {
@AllArgsConstructor
@Getter
static class Buffer {
ElasticIndex index;
String index;
Object payload;
}
private Mono<String> getIndexForSave(String index) {
return indexManager
.getIndexStrategy(index)
.map(strategy -> strategy.getIndexForSave(index));
}
private Mono<String> getIndexForSearch(String index) {
return indexManager
.getIndexStrategy(index)
.map(strategy -> strategy.getIndexForSearch(index));
}
protected Mono<Integer> doSave(Collection<Buffer> buffers) {
return Flux.fromIterable(buffers)
.collect(Collectors.groupingBy(Buffer::getIndex))
.map(Map::entrySet)
.flatMapIterable(Function.identity())
.flatMapMany(map -> Flux
.fromIterable(map.entrySet())
.flatMap(e ->
this.getIndexForSave(e.getKey())
.zipWith(indexManager.getIndexMetadata(e.getKey()), (index, metadata) -> Tuples.of(index, metadata, e.getValue()))))
.map(entry -> {
ElasticIndex index = entry.getKey();
BulkRequest bulkRequest = new BulkRequest(index.getStandardIndex(), index.getStandardType());
for (Buffer buffer : entry.getValue()) {
String index = entry.getT1();
BulkRequest bulkRequest = new BulkRequest(index, "_doc");
for (Buffer buffer : entry.getT3()) {
IndexRequest request = new IndexRequest();
Object o = JSON.toJSON(buffer.getPayload());
if (o instanceof Map) {
request.source((Map) o);
request.source(ElasticSearchConverter.convertDataToElastic((Map<String, Object>) o, entry.getT2().getProperties()));
} else {
request.source(o.toString(), XContentType.JSON);
}
bulkRequest.add(request);
}
entry.getValue().clear();
entry.getT3().clear();
return bulkRequest;
})
.flatMap(bulkRequest ->
Mono.<Integer>create(sink ->
Mono.create(sink ->
restClient.getWriteClient()
.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
@Override
@ -184,7 +202,7 @@ public class DefaultElasticSearchService implements ElasticSearchService {
sink.error(new RuntimeException("保存ElasticSearch数据失败:" + responses.buildFailureMessage()));
return;
}
sink.success(buffers.size());
sink.success(1);
}
@Override
@ -192,97 +210,68 @@ public class DefaultElasticSearchService implements ElasticSearchService {
sink.error(e);
}
})))
.collect(Collectors.summingInt(Integer::intValue));
.then(Mono.just(buffers.size()));
}
private <T> PagerResult<T> translatePageResult(Class<T> clazz, QueryParam param, SearchResponse response) {
private <T> PagerResult<T> translatePageResult(Function<Map<String, Object>, T> mapper, QueryParam param, SearchResponse response) {
long total = response.getHits().getTotalHits();
return PagerResult.of((int) total, translate(clazz, response), param);
return PagerResult.of((int) total, translate(mapper, response), param);
}
private <T> List<T> translate(Class<T> clazz, SearchResponse response) {
// TODO: 2020/1/18 临时代码
private <T> List<T> translate(Function<Map<String, Object>, T> mapper, SearchResponse response) {
return Arrays.stream(response.getHits().getHits())
.map(hit -> {
Map<String, Object> hitMap = hit.getSourceAsMap();
hitMap.put("id", hit.getId());
return JSON.toJavaObject(new JSONObject(hitMap), clazz);
return mapper.apply(hitMap);
})
.collect(Collectors.toList());
}
private Mono<SearchResponse> query(Mono<SearchRequest> requestMono) {
return requestMono.<SearchResponse>flatMap((request) ->
Mono.create(sink -> restClient
.getQueryClient()
.searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(sink))))
private Mono<SearchResponse> doSearch(Mono<SearchRequest> requestMono) {
return requestMono.flatMap((request) ->
ReactorActionListener
.<SearchResponse>mono(listener ->
restClient
.getQueryClient()
.searchAsync(request, RequestOptions.DEFAULT, listener)))
.onErrorResume(err -> {
log.error("query elastic error", err);
return Mono.empty();
});
}
private Mono<CountResponse> countQuery(Mono<CountRequest> requestMono) {
return requestMono.<CountResponse>flatMap((request) ->
Mono.create(sink -> restClient
.getQueryClient()
.countAsync(request, RequestOptions.DEFAULT, translatorActionListener(sink))))
private Mono<CountResponse> doCount(Mono<CountRequest> requestMono) {
return requestMono.flatMap((request) ->
ReactorActionListener
.<CountResponse>mono(listener ->
restClient
.getQueryClient()
.countAsync(request, RequestOptions.DEFAULT, listener)))
.onErrorResume(err -> {
log.error("query elastic error", err);
return Mono.empty();
});
}
private <T> ActionListener<T> translatorActionListener(MonoSink<T> sink) {
return new ActionListener<T>() {
@Override
public void onResponse(T response) {
sink.success(response);
}
@Override
public void onFailure(Exception e) {
if (e instanceof ElasticsearchException) {
if (((ElasticsearchException) e).status().getStatus() == 404) {
sink.success();
return;
} else if (((ElasticsearchException) e).status().getStatus() == 400) {
sink.error(new ElasticsearchParseException("查询参数格式错误", e));
}
}
sink.error(e);
}
};
private Mono<SearchRequest> createSearchRequest(QueryParam queryParam, String index) {
return indexManager
.getIndexMetadata(index)
.map(metadata -> ElasticSearchConverter.convertSearchSourceBuilder(queryParam, metadata))
.switchIfEmpty(Mono.fromSupplier(() -> ElasticSearchConverter.convertSearchSourceBuilder(queryParam, null)))
.flatMap(builder -> this.getIndexForSearch(index)
.map(idx -> new SearchRequest(idx).source(builder).types("_doc")));
}
private Mono<SearchRequest> searchRequestStructure(QueryParam queryParam, ElasticIndex provider) {
return indexOperationService.getIndexMappingMetadata(provider.getStandardIndex())
.switchIfEmpty(Mono.just(IndexMappingMetadata.getInstance(provider.getStandardIndex())))
.map(metadata -> {
SearchRequest request = new SearchRequest(provider.getStandardIndex())
.source(translateService.translate(queryParam, metadata));
if (StringUtils.hasText(provider.getStandardType())) {
request.types(provider.getStandardType());
}
return request;
})
.doOnNext(searchRequest -> log.debug("查询index{},es查询参数:{}", provider.getStandardIndex(), searchRequest.source().toString()))
.onErrorResume(err -> {
log.error("query index error", err);
return Mono.empty();
});
}
private Mono<CountRequest> countRequestStructure(QueryParam queryParam, ElasticIndex provider) {
private Mono<CountRequest> createCountRequest(QueryParam queryParam, String index) {
QueryParam tempQueryParam = queryParam.clone();
tempQueryParam.setPaging(false);
tempQueryParam.setSorts(Collections.emptyList());
return indexOperationService.getIndexMappingMetadata(provider.getStandardIndex())
.map(metadata -> new CountRequest(provider.getStandardIndex())
.source(translateService.translate(tempQueryParam, metadata)))
.onErrorResume(err -> {
log.error("query index error", err);
return Mono.empty();
});
return indexManager
.getIndexMetadata(index)
.map(metadata -> ElasticSearchConverter.convertSearchSourceBuilder(queryParam, metadata))
.switchIfEmpty(Mono.fromSupplier(() -> ElasticSearchConverter.convertSearchSourceBuilder(queryParam, null)))
.flatMap(builder -> this.getIndexForSearch(index)
.map(idx -> new CountRequest(idx).source(builder)));
}
}

View File

@ -2,25 +2,68 @@ package org.jetlinks.community.elastic.search.service;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.bean.FastBeanCopier;
import org.jetlinks.community.elastic.search.index.ElasticIndex;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Collection;
import java.util.Map;
import java.util.function.Function;
public interface ElasticSearchService {
<T> Mono<PagerResult<T>> queryPager(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper);
<T> Mono<PagerResult<T>> queryPager(ElasticIndex index, QueryParam queryParam, Class<T> type);
<T> Flux<T> query(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper);
<T> Flux<T> query(ElasticIndex index, QueryParam queryParam, Class<T> type);
Mono<Long> count(String index, QueryParam queryParam);
Mono<Long> count(ElasticIndex index, QueryParam queryParam);
<T> Mono<Void> commit(String index, T payload);
<T> Mono<Void> commit(ElasticIndex index, T payload);
<T> Mono<Void> commit(String index, Collection<T> payload);
<T> Mono<Void> commit(ElasticIndex index, Collection<T> payload);
<T> Mono<Void> commit(String index, Publisher<T> data);
default <T> Flux<T> query(String index, QueryParam queryParam, Class<T> type) {
return query(index, queryParam, map -> FastBeanCopier.copy(map, type));
}
default <T> Mono<PagerResult<T>> queryPager(String index, QueryParam queryParam, Class<T> type) {
return queryPager(index, queryParam, map -> FastBeanCopier.copy(map, type));
}
default <T> Mono<PagerResult<T>> queryPager(ElasticIndex index, QueryParam queryParam, Class<T> type) {
return queryPager(index.getIndex(), queryParam, map -> FastBeanCopier.copy(map, type));
}
default <T> Mono<PagerResult<T>> queryPager(ElasticIndex index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
return queryPager(index.getIndex(), queryParam, mapper);
}
default <T> Flux<T> query(ElasticIndex index, QueryParam queryParam, Class<T> type) {
return query(index.getIndex(), queryParam, map -> FastBeanCopier.copy(map, type));
}
default <T> Flux<T> query(ElasticIndex index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
return this.query(index.getIndex(), queryParam, mapper);
}
default <T> Mono<Long> count(ElasticIndex index, QueryParam data) {
return this.count(index.getIndex(), data);
}
default <T> Mono<Void> commit(ElasticIndex index, T data) {
return this.commit(index.getIndex(), data);
}
default <T> Mono<Void> commit(ElasticIndex index, Collection<T> data) {
return this.commit(index.getIndex(), data);
}
default <T> Mono<Void> commit(ElasticIndex index, Publisher<T> data) {
return this.commit(index.getIndex(), data);
}
<T> Mono<Void> commit(ElasticIndex index, Publisher<T> data);
}

View File

@ -1,15 +0,0 @@
package org.jetlinks.community.elastic.search.service;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import reactor.core.publisher.Mono;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexAliasOperationService {
Mono<Boolean> indexAliasIsExists(String index);
Mono<Boolean> AddAlias(IndicesAliasesRequest request);
}

View File

@ -1,18 +0,0 @@
package org.jetlinks.community.elastic.search.service;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
import reactor.core.publisher.Mono;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexOperationService {
Mono<Boolean> indexIsExists(String index);
Mono<Boolean> init(CreateIndexRequest request);
Mono<IndexMappingMetadata> getIndexMappingMetadata(String index);
}

View File

@ -1,15 +0,0 @@
package org.jetlinks.community.elastic.search.service;
import org.elasticsearch.client.indices.PutIndexTemplateRequest;
import reactor.core.publisher.Mono;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexTemplateOperationService {
Mono<Boolean> indexTemplateIsExists(String index);
Mono<Boolean> putTemplate(PutIndexTemplateRequest request);
}

View File

@ -1,20 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.community.timeseries.TimeSeriesService;
/**
* @author bsetfeng
* @since 1.0
**/
public abstract class AbstractTimeSeriesService implements TimeSeriesService {
protected QueryParam filterAddDefaultSort(QueryParam queryParam) {
if (queryParam.getSorts().isEmpty()) {
queryParam.orderBy("timestamp").desc();
}
return queryParam;
}
}

View File

@ -1,77 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.jetlinks.community.elastic.search.index.IndexTemplateProvider;
import org.jetlinks.community.elastic.search.manager.StandardsIndexManager;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.elastic.search.service.IndexTemplateOperationService;
import org.jetlinks.community.timeseries.TimeSeriesManager;
import org.jetlinks.community.timeseries.TimeSeriesMetric;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author bsetfeng
* @since 1.0
**/
@Getter
@Slf4j
public abstract class ESAbstractTimeSeriesManager implements TimeSeriesManager {
private Map<String, LocalTimeSeriesMetric> localMetricMap = new ConcurrentHashMap<>();
protected StandardsIndexManager standardsIndexManager;
protected IndexOperationService indexOperationService;
protected IndexTemplateOperationService indexTemplateOperationService;
public ESAbstractTimeSeriesManager(IndexOperationService indexOperationService,
StandardsIndexManager standardsIndexManager,
IndexTemplateOperationService indexTemplateOperationService) {
this.indexOperationService = indexOperationService;
this.standardsIndexManager = standardsIndexManager;
this.indexTemplateOperationService = indexTemplateOperationService;
}
// TODO: 2020/2/11 策略变更动态更新实现
protected LocalTimeSeriesMetric getLocalTimeSeriesMetric(TimeSeriesMetric metric) {
return localMetricMap.computeIfAbsent(metric.getId().toLowerCase(), index -> {
String standardsIndex = getStandardsIndexManager().getStandardsIndex(index);
String templateName = IndexTemplateProvider.getIndexTemplate(index);
return new LocalTimeSeriesMetric(
getStandardsIndexManager().indexIsChange(index),
index,
standardsIndex,
IndexAliasProvider.getIndexAlias(index),
templateName,
Collections.singletonList(index.concat("*")));
});
}
@Getter
@AllArgsConstructor
static class LocalTimeSeriesMetric {
private boolean indexHasStrategy;
private String index;
private String standardsIndex;
private Alias alias;
private String templateName;
private List<String> indexTemplatePatterns;
}
}

View File

@ -1,24 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.jetlinks.community.timeseries.query.AggregationData;
import java.util.Map;
/**
* @author bsetfeng
* @since 1.0
**/
@Getter
@AllArgsConstructor
public class ESAggregationData implements AggregationData {
private Map<String, Object> map;
@Override
public Map<String, Object> asMap() {
return this.map;
}
}

View File

@ -1,100 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.indices.PutIndexTemplateRequest;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import org.jetlinks.community.elastic.search.index.mapping.MappingFactory;
import org.jetlinks.community.elastic.search.manager.StandardsIndexManager;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.elastic.search.service.IndexTemplateOperationService;
import org.jetlinks.community.timeseries.TimeSeriesMetadata;
import org.jetlinks.community.timeseries.TimeSeriesMetric;
import org.jetlinks.community.timeseries.TimeSeriesService;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.util.Collections;
import java.util.List;
/**
* @author bsetfeng
* @since 1.0
**/
@Service
@Slf4j
public class ESTimeSeriesManager extends ESAbstractTimeSeriesManager {
private final TimeSeriesServiceRegisterCenter timeSeriesServiceRegisterCenter;
public ESTimeSeriesManager(TimeSeriesServiceRegisterCenter timeSeriesServiceRegisterCenter,
IndexOperationService indexOperationService,
StandardsIndexManager standardsIndexManager,
IndexTemplateOperationService indexTemplateOperationService) {
super(indexOperationService,standardsIndexManager,indexTemplateOperationService);
this.timeSeriesServiceRegisterCenter = timeSeriesServiceRegisterCenter;
}
@Override
public TimeSeriesService getService(TimeSeriesMetric metric) {
return timeSeriesServiceRegisterCenter.getTimeSeriesService(getLocalTimeSeriesMetric(metric));
}
@Override
public Mono<Void> registerMetadata(TimeSeriesMetadata metadata) {
LocalTimeSeriesMetric localTimeSeriesMetric = getLocalTimeSeriesMetric(metadata.getMetric());
return registerIndexTemplate(localTimeSeriesMetric, metadata.getProperties())
.doOnError(e -> log.error("初始化模板:{}失败", localTimeSeriesMetric.getTemplateName(), e))
.then();
}
private Mono<Boolean> registerIndexTemplate(LocalTimeSeriesMetric localTimeSeriesMetric, List<PropertyMetadata> properties) {
return Mono.<PutIndexTemplateRequest>create(sink -> {
MappingFactory factory = CreateIndex.createInstance().createMapping();
properties
.forEach(propertyMetadata -> addField(propertyMetadata, factory));
addDefaultProperty(factory);
PutIndexTemplateRequest request = new PutIndexTemplateRequest(localTimeSeriesMetric.getTemplateName());
request.alias(localTimeSeriesMetric.getAlias());
request.mapping(Collections.singletonMap("properties", factory.end().getMapping()));
request.patterns(localTimeSeriesMetric.getIndexTemplatePatterns());
sink.success(request);
})
.flatMap(indexTemplateOperationService::putTemplate)
.doOnError(e -> log.error("初始化template:{}失败", localTimeSeriesMetric.getTemplateName(), e));
}
private void addDefaultProperty(MappingFactory factory) {
factory
.addFieldName("timestamp")
.addFieldType(FieldType.DATE)
.addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time)
.commit();
}
private void addField(PropertyMetadata property, MappingFactory factory) {
// TODO: 2020/2/4 对object 类型的支持
DataType type = property.getValueType();
if (type.getType().equalsIgnoreCase("date")) {
DateTimeType dateTimeType = (DateTimeType) type;
factory
.addFieldName(property.getId())
.addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time)
//.addFieldDateFormat(dateTimeType.getFormat())
.addFieldType(FieldType.DATE)
.commit();
} else {
factory
.addFieldName(property.getId())
.addFieldType(FieldType.ofJava(type.getType()))
.commit();
}
}
}

View File

@ -1,241 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.AllArgsConstructor;
import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.core.param.TermType;
import org.jetlinks.community.elastic.search.aggreation.bucket.Bucket;
import org.jetlinks.community.elastic.search.aggreation.bucket.BucketAggregationsStructure;
import org.jetlinks.community.elastic.search.aggreation.bucket.Sort;
import org.jetlinks.community.elastic.search.aggreation.enums.BucketType;
import org.jetlinks.community.elastic.search.aggreation.enums.MetricsType;
import org.jetlinks.community.elastic.search.aggreation.enums.OrderType;
import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsAggregationStructure;
import org.jetlinks.community.elastic.search.index.ElasticIndex;
import org.jetlinks.community.elastic.search.service.AggregationService;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.timeseries.TimeSeriesData;
import org.jetlinks.community.timeseries.query.AggregationData;
import org.jetlinks.community.timeseries.query.AggregationQueryParam;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;
/**
* @author bsetfeng
* @since 1.0
**/
@AllArgsConstructor
public class ESTimeSeriesService extends AbstractTimeSeriesService {
private ESAbstractTimeSeriesManager.LocalTimeSeriesMetric localTimeSeriesMetric;
private ElasticSearchService elasticSearchService;
private AggregationService aggregationService;
@Override
public Flux<TimeSeriesData> query(QueryParam queryParam) {
return elasticSearchService.query(
cleverGetIndex(),
filterAddDefaultSort(queryParam),
Map.class)
.map(map -> new TimeSeriesData() {
@Override
public long getTimestamp() {
return (long) map.get("timestamp");
}
@Override
public Map<String, Object> getData() {
return map;
}
});
}
@Override
public Mono<Integer> count(QueryParam queryParam) {
return elasticSearchService.count(
cleverGetIndex(),
queryParam
).map(Long::intValue);
}
@Override
public Flux<AggregationData> aggregation(AggregationQueryParam param) {
BucketAggregationsStructure structure = getStandardStructure(param);
//由于es一次性返回所有聚合查询结果不完全支持响应式规范
return aggregationService.bucketAggregation(
filterAddDefaultSort(addQueryTimeRange(param.getQueryParam(), param)),
structure,
cleverGetIndex())
.onErrorResume(err -> Mono.empty())
.flatMapMany(bucketResponse -> Flux.fromIterable(new BucketsParser(bucketResponse.getBuckets()).result)
.take(param.getLimit())
.map(ESAggregationData::new))
;
}
@Override
public Mono<Void> save(Publisher<TimeSeriesData> data) {
return Flux.from(data)
.map(timeSeriesData -> {
Map<String, Object> map = timeSeriesData.getData();
map.put("timestamp", timeSeriesData.getTimestamp());
return map;
})
.as(stream -> elasticSearchService.commit(getDefaultIndex(), stream));
}
@Override
public Mono<Void> save(TimeSeriesData data) {
Map<String, Object> map = data.getData();
map.put("timestamp", data.getTimestamp());
return elasticSearchService.commit(getDefaultIndex(), map);
}
public void pack(BucketAggregationsStructure structureOutLayer, BucketAggregationsStructure structureInnerLayer) {
structureOutLayer.setSubBucketAggregation(Collections.singletonList(structureInnerLayer));
}
protected BucketAggregationsStructure getStandardStructure(AggregationQueryParam param) {
List<BucketAggregationsStructure> structures = new ArrayList<>();
if (param.getGroupByTime() != null) {
structures.add(getDateTypeStructure(param));
}
if (param.getGroupBy() != null && !param.getGroupBy().isEmpty()) {
structures.addAll(getTermTypeStructures(param));
}
for (int i = 0; i < structures.size(); i++) {
if (i < structures.size() - 1) {
pack(structures.get(i), structures.get(i + 1));
}
if (i == structures.size() - 1) {
structures.get(i).setSubMetricsAggregation(param.getAggColumns()
.stream()
.map(agg -> {
MetricsAggregationStructure metricsAggregationStructure = new MetricsAggregationStructure();
metricsAggregationStructure.setField(agg.getProperty());
metricsAggregationStructure.setName(agg.getAlias());
metricsAggregationStructure.setType(MetricsType.of(agg.getAggregation().name()));
return metricsAggregationStructure;
}).collect(Collectors.toList()));
}
}
return structures.get(0);
}
protected BucketAggregationsStructure getDateTypeStructure(AggregationQueryParam param) {
BucketAggregationsStructure structure = new BucketAggregationsStructure();
structure.setInterval(durationFormat(param.getGroupByTime().getInterval()));
structure.setType(BucketType.DATE_HISTOGRAM);
structure.setFormat(param.getGroupByTime().getFormat());
structure.setName(param.getGroupByTime().getAlias());
structure.setField("timestamp");
structure.setSort(Sort.desc(OrderType.KEY));
structure.setExtendedBounds(getExtendedBounds(param));
return structure;
}
protected List<BucketAggregationsStructure> getTermTypeStructures(AggregationQueryParam param) {
return param.getGroupBy()
.stream()
.map(group -> {
BucketAggregationsStructure structure = new BucketAggregationsStructure();
structure.setType(BucketType.TERMS);
structure.setSize(param.getLimit());
structure.setField(group.getProperty());
structure.setName(group.getAlias());
return structure;
}).collect(Collectors.toList());
}
protected ElasticIndex getDefaultIndex() {
return ElasticIndex.createDefaultIndex(() -> localTimeSeriesMetric.getStandardsIndex(), () -> "_doc");
}
protected ElasticIndex cleverGetIndex() {
if (localTimeSeriesMetric.isIndexHasStrategy()) {
return ElasticIndex.createDefaultIndex(() -> localTimeSeriesMetric.getAlias().name(), () -> "_doc");
} else {
return ElasticIndex.createDefaultIndex(() -> localTimeSeriesMetric.getIndex(), () -> "_doc");
}
}
protected static QueryParam addQueryTimeRange(QueryParam queryParam, AggregationQueryParam param) {
queryParam.and(
"timestamp",
TermType.btw,
Arrays.asList(getStartWithParam(param), param.getEndWithTime()));
return queryParam;
}
protected static ExtendedBounds getExtendedBounds(AggregationQueryParam param) {
return new ExtendedBounds(getStartWithParam(param), param.getEndWithTime());
}
private static long getStartWithParam(AggregationQueryParam param) {
long startWithParam = param.getStartWithTime();
if (param.getGroupByTime() != null && param.getGroupByTime().getInterval() != null) {
long timeInterval = param.getGroupByTime().getInterval().toMillis() * param.getLimit();
long tempStartWithParam = param.getEndWithTime() - timeInterval;
startWithParam = Math.max(tempStartWithParam, startWithParam);
}
return startWithParam;
}
protected static String durationFormat(Duration duration) {
String durationStr = duration.toString();
if (durationStr.contains("S")) {
return duration.toMillis() / 1000 + "s";
} else if (!durationStr.contains("S") && durationStr.contains("M")) {
return duration.toMinutes() + "m";
} else if (!durationStr.contains("S") && !durationStr.contains("M")) {
if (duration.toHours() % 24 == 0) {
return duration.toDays() + "d";
} else {
return duration.toHours() + "h";
}
}
throw new UnsupportedOperationException("不支持的格式化Duration: " + duration.toString());
}
static class BucketsParser {
private List<Map<String, Object>> result;
public BucketsParser(List<Bucket> buckets) {
result = new ArrayList<>();
buckets.forEach(bucket -> parser(bucket, new HashMap<>()));
}
public void parser(Bucket bucket, Map<String, Object> fMap) {
addBucketProperty(bucket, fMap);
if (bucket.getBuckets() != null && !bucket.getBuckets().isEmpty()) {
bucket.getBuckets().forEach(b -> {
Map<String, Object> map = new HashMap<>(fMap);
addBucketProperty(b, map);
parser(b, map);
});
} else {
result.add(fMap);
}
}
private void addBucketProperty(Bucket bucket, Map<String, Object> fMap) {
fMap.put(bucket.getName(), bucket.getKey());
fMap.putAll(bucket.toMap());
}
}
}

View File

@ -0,0 +1,69 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.metadata.SimplePropertyMetadata;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.elastic.search.service.AggregationService;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.timeseries.TimeSeriesManager;
import org.jetlinks.community.timeseries.TimeSeriesMetadata;
import org.jetlinks.community.timeseries.TimeSeriesMetric;
import org.jetlinks.community.timeseries.TimeSeriesService;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author bsetfeng
* @author zhouhao
* @since 1.0
**/
@Service
@Slf4j
public class ElasticSearchTimeSeriesManager implements TimeSeriesManager {
private Map<String, TimeSeriesService> serviceMap = new ConcurrentHashMap<>(16);
protected final ElasticSearchIndexManager indexManager;
private final ElasticSearchService elasticSearchService;
private final AggregationService aggregationService;
public ElasticSearchTimeSeriesManager(ElasticSearchIndexManager indexManager,
ElasticSearchService elasticSearchService,
AggregationService aggregationService) {
this.elasticSearchService = elasticSearchService;
this.indexManager = indexManager;
this.aggregationService = aggregationService;
}
@Override
public TimeSeriesService getService(TimeSeriesMetric metric) {
return getService(metric.getId());
}
@Override
public TimeSeriesService getService(String metric) {
return serviceMap.computeIfAbsent(metric,
id -> new ElasticSearchTimeSeriesService(id, elasticSearchService, aggregationService));
}
@Override
public Mono<Void> registerMetadata(TimeSeriesMetadata metadata) {
//默认字段
SimplePropertyMetadata timestamp = new SimplePropertyMetadata();
timestamp.setId("timestamp");
timestamp.setValueType(new DateTimeType());
return indexManager.putIndex(new DefaultElasticSearchIndexMetadata(metadata.getMetric().getId(), metadata.getProperties())
.addProperty(timestamp));
}
}

View File

@ -0,0 +1,68 @@
package org.jetlinks.community.elastic.search.timeseries;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.community.elastic.search.service.AggregationService;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.timeseries.TimeSeriesData;
import org.jetlinks.community.timeseries.TimeSeriesService;
import org.jetlinks.community.timeseries.query.AggregationData;
import org.jetlinks.community.timeseries.query.AggregationQueryParam;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Map;
@AllArgsConstructor
@Slf4j
public class ElasticSearchTimeSeriesService implements TimeSeriesService {
private String index;
private ElasticSearchService elasticSearchService;
private AggregationService aggregationService;
@Override
public Flux<TimeSeriesData> query(QueryParam queryParam) {
return elasticSearchService.query(index, queryParam, map -> TimeSeriesData.of((Long) map.get("timestamp"), map));
}
@Override
public Mono<Integer> count(QueryParam queryParam) {
return elasticSearchService
.count(index, queryParam)
.map(Long::intValue);
}
@Override
public Flux<AggregationData> aggregation(AggregationQueryParam queryParam) {
return aggregationService
.aggregation(index, queryParam)
.onErrorResume(err -> {
log.error(err.getMessage(), err);
return Mono.empty();
})
.map(AggregationData::of);
}
@Override
public Mono<Void> save(Publisher<TimeSeriesData> data) {
return Flux.from(data)
.flatMap(this::save)
.then();
}
@Override
public Mono<Void> save(TimeSeriesData data) {
return Mono.defer(() -> {
Map<String, Object> mapData = data.getData();
mapData.put("timestamp", data.getTimestamp());
return elasticSearchService.commit(index, mapData);
});
}
}

View File

@ -1,14 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import org.elasticsearch.action.admin.indices.alias.Alias;
/**
* @author bsetfeng
* @since 1.0
**/
public interface IndexAliasProvider {
static Alias getIndexAlias(String index) {
return new Alias(index.concat("_alias"));
}
}

View File

@ -1,32 +0,0 @@
package org.jetlinks.community.elastic.search.timeseries;
import org.jetlinks.community.elastic.search.service.AggregationService;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.timeseries.TimeSeriesService;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author bsetfeng
* @since 1.0
**/
@Component
public class TimeSeriesServiceRegisterCenter {
private final ElasticSearchService elasticSearchService;
private final AggregationService aggregationService;
private Map<String, TimeSeriesService> serviceMap = new ConcurrentHashMap<>(16);
public TimeSeriesServiceRegisterCenter(AggregationService aggregationService, ElasticSearchService elasticSearchService) {
this.aggregationService = aggregationService;
this.elasticSearchService = elasticSearchService;
}
public TimeSeriesService getTimeSeriesService(ESAbstractTimeSeriesManager.LocalTimeSeriesMetric metric) {
return serviceMap.computeIfAbsent(metric.getIndex(), i -> new ESTimeSeriesService(metric, elasticSearchService, aggregationService));
}
}

View File

@ -1,44 +0,0 @@
package org.jetlinks.community.elastic.search.translate;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.community.elastic.search.enums.LinkTypeEnum;
import java.util.Objects;
/**
* @author bsetfeng
* @since 1.0
**/
@Slf4j
public class QueryParamTranslator {
public static QueryBuilder translate(QueryParam queryParam) {
BoolQueryBuilder query = QueryBuilders.boolQuery();
Objects.requireNonNull(queryParam, "QueryParam must not null.")
.getTerms()
.forEach(term -> LinkTypeEnum.of(term.getType().name())
.ifPresent(e -> e.process(query, term)));
return query;
}
// public static SearchSourceBuilder transSourceBuilder(QueryParam queryParam) {
// SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
// if (queryParam.isPaging()) {
// sourceBuilder.from(queryParam.getPageIndex() * queryParam.getPageSize());
// sourceBuilder.size(queryParam.getPageSize());
// }
// queryParam.getSorts()
// .forEach(sort -> {
// if (!StringUtils.isEmpty(sort.getName())) {
// sourceBuilder.sort(sort.getName(), SortOrder.fromString(sort.getOrder()));
// }
//
// });
// return sourceBuilder.query(translate(queryParam));
// }
}

View File

@ -1,54 +0,0 @@
package org.jetlinks.community.elastic.search.utils;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
/**
* @author bsetfeng
* @since 1.0
**/
@Slf4j
public class DateTimeUtils {
public static Object formatDateToTimestamp(Object date, List<String> formats) {
return TermCommonUtils.getStandardsTermValue(
formatDateArrayToTimestamp(TermCommonUtils.convertToList(date), formats)
);
}
private static Object formatDateStringToTimestamp(String dateString, List<String> formats) {
for (String format : formats) {
try {
return formatDateStringToTimestamp(dateString, format);
} catch (Exception e) {
log.debug("按格式:{}解析时间字符串:{}错误", format, dateString);
}
}
throw new UnsupportedOperationException("不支持的时间转换" + "formats:" +
JSON.toJSONString(formats) + "dateString:" + dateString);
}
private static long formatDateStringToTimestamp(String dateString, String format) {
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(format);
LocalDateTime dateTime = LocalDateTime.parse(dateString, dateTimeFormatter);
return dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
}
private static List<Object> formatDateArrayToTimestamp(List<Object> values, List<String> formats) {
List<Object> result = new ArrayList<>();
for (Object value : values) {
if (value instanceof String) {
result.add(formatDateStringToTimestamp(value.toString(), formats));
} else {
result.add(value);
}
}
return result;
}
}

View File

@ -0,0 +1,63 @@
package org.jetlinks.community.elastic.search.utils;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.jetlinks.core.metadata.Converter;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.core.metadata.types.GeoPoint;
import org.jetlinks.core.metadata.types.GeoType;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class ElasticSearchConverter {
public static SearchSourceBuilder convertSearchSourceBuilder(QueryParam queryParam, ElasticSearchIndexMetadata metadata) {
return QueryParamTranslator.convertSearchSourceBuilder(queryParam, metadata);
}
public static Map<String, Object> convertDataToElastic(Map<String, Object> data, List<PropertyMetadata> properties) {
for (PropertyMetadata property : properties) {
DataType type = property.getValueType();
Object val = data.get(property.getId());
if (val == null) {
continue;
}
//处理地理位置类型
if (type instanceof GeoType) {
GeoPoint point = ((GeoType) type).convert(val);
Map<String, Object> geoData = new HashMap<>();
geoData.put("lat", point.getLat());
geoData.put("lon", point.getLon());
data.put(property.getId(), geoData);
}if (type instanceof DateTimeType) {
Date point = ((DateTimeType) type).convert(val);
data.put(property.getId(), point.getTime());
} else if (type instanceof Converter) {
data.put(property.getId(), ((Converter) type).convert(val));
}
}
return data;
}
public static Map<String, Object> convertDataFromElastic(Map<String, Object> data, List<PropertyMetadata> properties) {
for (PropertyMetadata property : properties) {
DataType type = property.getValueType();
Object val = data.get(property.getId());
if (val == null) {
continue;
}
//处理地理位置类型
if (type instanceof GeoType) {
data.put(property.getId(), ((GeoType) type).convertToMap(val));
}
}
return data;
}
}

View File

@ -0,0 +1,81 @@
package org.jetlinks.community.elastic.search.utils;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.core.param.Sort;
import org.hswebframework.ezorm.core.param.Term;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.types.GeoType;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.parser.DefaultLinkTypeParser;
import org.springframework.util.StringUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
/**
* @author zhouhao
* @since 1.0
**/
@Slf4j
public class QueryParamTranslator {
static DefaultLinkTypeParser linkTypeParser = new DefaultLinkTypeParser();
static Consumer<Term> doNotingParamConverter = (term -> {
});
static Map<String, BiConsumer<DataType, Term>> converter = new ConcurrentHashMap<>();
static BiConsumer<DataType, Term> defaultDataTypeConverter = (type, term) -> {
};
static {
//地理位置查询
converter.put(GeoType.ID, (type, term) -> {
// TODO: 2020/3/5
});
}
public static SearchSourceBuilder convertSearchSourceBuilder(QueryParam queryParam, ElasticSearchIndexMetadata metadata) {
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
if (queryParam.isPaging()) {
sourceBuilder.from(queryParam.getPageIndex() * queryParam.getPageSize());
sourceBuilder.size(queryParam.getPageSize());
}
for (Sort sort : queryParam.getSorts()) {
if (!StringUtils.isEmpty(sort.getName())) {
sourceBuilder.sort(sort.getName(), SortOrder.fromString(sort.getOrder()));
}
}
BoolQueryBuilder queryBuilders = QueryBuilders.boolQuery();
Consumer<Term> paramConverter = doNotingParamConverter;
if (metadata != null) {
paramConverter = t -> {
if (StringUtils.isEmpty(t.getColumn())) {
return;
}
PropertyMetadata property = metadata.getProperty(t.getColumn());
if (null != property) {
DataType type = property.getValueType();
converter.getOrDefault(type.getId(), defaultDataTypeConverter).accept(type, t);
}
};
}
for (Term term : queryParam.getTerms()) {
linkTypeParser.process(term, paramConverter, queryBuilders);
}
return sourceBuilder.query(queryBuilders);
}
}

View File

@ -0,0 +1,56 @@
package org.jetlinks.community.elastic.search.utils;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionListener;
import org.hswebframework.web.exception.BusinessException;
import reactor.core.publisher.Mono;
import java.util.function.Consumer;
import java.util.function.Function;
public class ReactorActionListener {
public static <R, T> Mono<R> mono(Consumer<ActionListener<T>> listenerConsumer,
Function<T, Mono<R>> onSuccess,
Function<Exception, Mono<R>> onError) {
return Mono.<Mono<R>>create(sink -> {
listenerConsumer.accept(new ActionListener<T>() {
@Override
public void onResponse(T t) {
try {
sink.success(onSuccess.apply(t));
} catch (Exception e) {
sink.error(e);
}
}
@Override
public void onFailure(Exception e) {
try {
sink.success(onError.apply(e));
} catch (Exception e2) {
sink.error(e2);
}
}
});
}).flatMap(Function.identity())
.onErrorResume(ElasticsearchStatusException.class, e -> {
if (e.status().getStatus() == 404) {
return Mono.empty();
}
return Mono.error(new BusinessException(e.getMessage(), e));
});
}
public static <R, T> Mono<R> mono(Consumer<ActionListener<T>> listenerConsumer,
Function<T, Mono<R>> onSuccess) {
return mono(listenerConsumer, onSuccess, Mono::error);
}
public static <R> Mono<R> mono(Consumer<ActionListener<R>> listenerConsumer) {
return mono(listenerConsumer, Mono::justOrEmpty, Mono::error);
}
}

View File

@ -24,7 +24,7 @@ public class TermCommonUtils {
return new ArrayList<Object>(((Collection) value));
}
return Arrays.asList(value);
return Collections.singletonList(value);
}
public static Object getStandardsTermValue(List<Object> value) {

View File

@ -1,19 +1,11 @@
package org.jetlinks.community.logging.configuration;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.logging.event.handler.LoggerIndexProvider;
import org.jetlinks.community.logging.logback.SystemLoggingAppender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.bind.annotation.PostMapping;
@Configuration
@EnableConfigurationProperties(LoggingProperties.class)

View File

@ -1,12 +1,12 @@
package org.jetlinks.community.logging.event.handler;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.core.metadata.types.ObjectType;
import org.jetlinks.core.metadata.types.StringType;
import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.logging.access.SerializableAccessLog;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
@ -24,23 +24,23 @@ public class AccessLoggerEventHandler {
private final ElasticSearchService elasticSearchService;
public AccessLoggerEventHandler(ElasticSearchService elasticSearchService, IndexOperationService indexOperationService) {
public AccessLoggerEventHandler(ElasticSearchService elasticSearchService, ElasticSearchIndexManager indexManager) {
this.elasticSearchService = elasticSearchService;
CreateIndexRequest accessLoggerIndex = CreateIndex.createInstance()
.addIndex(LoggerIndexProvider.ACCESS.getStandardIndex())
.createMapping()
.addFieldName("requestTime").addFieldType(FieldType.DATE).addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time).commit()
.addFieldName("responseTime").addFieldType(FieldType.DATE).addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time).commit()
.addFieldName("action").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("ip").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("url").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("httpHeaders").addFieldType(FieldType.OBJECT).commit()
.addFieldName("context").addFieldType(FieldType.OBJECT).commit()
.end()
.createIndexRequest();
indexOperationService.init(accessLoggerIndex)
.doOnError(err -> log.error(err.getMessage(), err))
.subscribe();
indexManager.putIndex(
new DefaultElasticSearchIndexMetadata(LoggerIndexProvider.ACCESS.getIndex())
.addProperty("requestTime", new DateTimeType())
.addProperty("responseTime", new DateTimeType())
.addProperty("action", new StringType())
.addProperty("ip", new StringType())
.addProperty("url", new StringType())
.addProperty("httpHeaders", new ObjectType())
.addProperty("context", new ObjectType()
.addProperty("userId",new StringType())
.addProperty("username",new StringType())
)
).subscribe();
}

View File

@ -12,10 +12,8 @@ import org.jetlinks.community.elastic.search.index.ElasticIndex;
@AllArgsConstructor
public enum LoggerIndexProvider implements ElasticIndex {
ACCESS("access_log", "_doc"),
SYSTEM("system_log", "_doc");
ACCESS("access_logger"),
SYSTEM("system_logger");
private String index;
private String type;
}

View File

@ -1,12 +1,12 @@
package org.jetlinks.community.logging.event.handler;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.core.metadata.types.ObjectType;
import org.jetlinks.core.metadata.types.StringType;
import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.elastic.search.service.ElasticSearchService;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.logging.system.SerializableSystemLog;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
@ -25,22 +25,19 @@ public class SystemLoggerEventHandler {
private final ElasticSearchService elasticSearchService;
public SystemLoggerEventHandler(ElasticSearchService elasticSearchService, IndexOperationService indexOperationService) {
public SystemLoggerEventHandler(ElasticSearchService elasticSearchService, ElasticSearchIndexManager indexManager) {
this.elasticSearchService = elasticSearchService;
CreateIndexRequest systemLoggerIndex = CreateIndex.createInstance()
.addIndex(LoggerIndexProvider.SYSTEM.getStandardIndex())
.createMapping()
.addFieldName("createTime").addFieldType(FieldType.DATE).addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time).commit()
.addFieldName("name").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("level").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("message").addFieldType(FieldType.KEYWORD).commit()
.end()
.createIndexRequest();
indexOperationService.init(systemLoggerIndex)
.doOnError(err -> log.error(err.getMessage(), err))
.subscribe();
indexManager.putIndex(
new DefaultElasticSearchIndexMetadata(LoggerIndexProvider.SYSTEM.getIndex())
.addProperty("createTime", new DateTimeType())
.addProperty("name", new StringType())
.addProperty("level", new StringType())
.addProperty("message", new StringType())
.addProperty("context", new ObjectType()
.addProperty("requestId",new StringType())
.addProperty("server",new StringType()))
).subscribe();
}
@EventListener

View File

@ -1,13 +1,11 @@
package org.jetlinks.community.rule.engine.configuration;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
import org.jetlinks.community.elastic.search.enums.FieldType;
import org.jetlinks.community.elastic.search.index.CreateIndex;
import org.jetlinks.community.elastic.search.service.IndexOperationService;
import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
import org.jetlinks.community.rule.engine.event.handler.RuleEngineLoggerIndexProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.jetlinks.core.metadata.types.DateTimeType;
import org.jetlinks.core.metadata.types.StringType;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
@ -18,40 +16,23 @@ import org.springframework.stereotype.Component;
@Component
@Order(1)
@Slf4j
public class RuleEngineLogIndexInitialize implements CommandLineRunner {
public class RuleEngineLogIndexInitialize {
private final IndexOperationService indexOperationService;
@Autowired
public RuleEngineLogIndexInitialize(IndexOperationService indexOperationService) {
this.indexOperationService = indexOperationService;
}
@Override
public void run(String... args) throws Exception {
indexOperationService.init(CreateIndex.createInstance()
.addIndex(RuleEngineLoggerIndexProvider.RULE_LOG.getStandardIndex())
.createMapping()
.addFieldName("createTime").addFieldType(FieldType.DATE).addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time).commit()
.addFieldName("level").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("message").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("nodeId").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("instanceId").addFieldType(FieldType.KEYWORD).commit()
.end()
.createIndexRequest())
.and(
indexOperationService.init(CreateIndex.createInstance()
.addIndex(RuleEngineLoggerIndexProvider.RULE_EVENT_LOG.getStandardIndex())
.createMapping()
.addFieldName("createTime").addFieldType(FieldType.DATE).addFieldDateFormat(FieldDateFormat.epoch_millis, FieldDateFormat.simple_date, FieldDateFormat.strict_date_time).commit()
.addFieldName("event").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("nodeId").addFieldType(FieldType.KEYWORD).commit()
.addFieldName("instanceId").addFieldType(FieldType.KEYWORD).commit()
.end()
.createIndexRequest())
public RuleEngineLogIndexInitialize(ElasticSearchIndexManager indexManager) {
indexManager.putIndex(new DefaultElasticSearchIndexMetadata(RuleEngineLoggerIndexProvider.RULE_LOG.getIndex())
.addProperty("createTime", new DateTimeType())
.addProperty("level", new StringType())
.addProperty("message", new StringType())
.addProperty("nodeId", new StringType())
.addProperty("instanceId", new StringType()))
.then(
indexManager.putIndex(new DefaultElasticSearchIndexMetadata(RuleEngineLoggerIndexProvider.RULE_LOG.getIndex())
.addProperty("createTime", new DateTimeType())
.addProperty("event", new StringType())
.addProperty("nodeId", new StringType())
.addProperty("instanceId", new StringType()))
)
.doOnError(err -> log.error(err.getMessage(), err))
.subscribe();
}
}

View File

@ -12,7 +12,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
@Component
@Slf4j
@ -26,7 +25,7 @@ public class RuleLogHandler {
@EventListener
public void handleRuleLog(LogInfo event) {
RuleEngineExecuteLogInfo logInfo = FastBeanCopier.copy(event, new RuleEngineExecuteLogInfo());
elasticSearchService.commit(RuleEngineLoggerIndexProvider.RULE_LOG, Mono.just(logInfo))
elasticSearchService.commit(RuleEngineLoggerIndexProvider.RULE_LOG, logInfo)
.subscribe();
}
@ -36,7 +35,7 @@ public class RuleLogHandler {
if (!RuleEvent.NODE_EXECUTE_BEFORE.equals(event.getEvent())
&& !RuleEvent.NODE_EXECUTE_RESULT.equals(event.getEvent())) {
RuleEngineExecuteEventInfo eventInfo = FastBeanCopier.copy(event, new RuleEngineExecuteEventInfo());
elasticSearchService.commit(RuleEngineLoggerIndexProvider.RULE_EVENT_LOG, Mono.just(eventInfo))
elasticSearchService.commit(RuleEngineLoggerIndexProvider.RULE_EVENT_LOG, eventInfo)
.subscribe();
}
}

View File

@ -18,6 +18,8 @@ public interface TimeSeriesManager {
*/
TimeSeriesService getService(TimeSeriesMetric metric);
TimeSeriesService getService(String metric);
/**
* 注册元数据
*

View File

@ -14,4 +14,7 @@ public interface AggregationData extends ValueObject {
return Optional.ofNullable(asMap().get(name));
}
static AggregationData of(Map<String,Object> map){
return ()->map;
}
}

View File

@ -6,12 +6,9 @@ import org.hswebframework.ezorm.core.dsl.Query;
import org.hswebframework.ezorm.core.param.QueryParam;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;
@ -35,6 +32,8 @@ public class AggregationQueryParam {
private long endWithTime = System.currentTimeMillis();
private String timeProperty = "timestamp";
//条件过滤
private QueryParam queryParam = new QueryParam();

View File

@ -13,6 +13,8 @@ import java.time.Duration;
@NoArgsConstructor
public class TimeGroup {
private String property;
//时间分组间隔,: 1d , 30s
private Duration interval;
@ -23,5 +25,9 @@ public class TimeGroup {
*/
private String format;
public TimeGroup(Duration interval, String alias, String format) {
this.interval = interval;
this.alias = alias;
this.format = format;
}
}

View File

@ -40,6 +40,8 @@ elasticsearch:
connect-timeout: 5000
socket-timeout: 5000
connection-request-timeout: 8000
index:
default-strategy: time-by-month #默认es的索引按月进行分表, direct则为直接操作索引.
device:
message:
writer: