一大波优化

jetlinks-components/common-component/pom.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>jetlinks-components</artifactId>
+        <groupId>org.jetlinks.community</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>common-component</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.jetlinks</groupId>
+            <artifactId>jetlinks-core</artifactId>
+            <version>${jetlinks.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-authorization-api</artifactId>
+            <version>${hsweb.framework.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-starter</artifactId>
+            <version>${hsweb.framework.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+        </dependency>
+    </dependencies>
+</project>

jetlinks-components/common-component/src/main/java/org/jetlinks/community/ConfigMetadataConstants.java

@@ -0,0 +1,19 @@
+package org.jetlinks.community;
+
+import org.jetlinks.core.config.ConfigKey;
+
+/**
+ * @see ConfigKey
+ */
+public interface ConfigMetadataConstants {
+
+    //字符串相关配置
+    ConfigKey<Long> maxLength = ConfigKey.of("maxLength", "字符串最大长度", Long.TYPE);
+    ConfigKey<Boolean> isRichText = ConfigKey.of("isRichText", "是否为富文本", Boolean.TYPE);
+    ConfigKey<Boolean> isScript = ConfigKey.of("isScript", "是否为脚本", Boolean.TYPE);
+
+    ConfigKey<Boolean> allowInput = ConfigKey.of("allowInput", "允许输入", Boolean.TYPE);
+    ConfigKey<Boolean> required = ConfigKey.of("required", "是否必填", Boolean.TYPE);
+
+
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/ValueObject.java

@@ -0,0 +1,48 @@
+package org.jetlinks.community;
+
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.jetlinks.community.utils.TimeUtils;
+
+import java.time.Duration;
+import java.util.Date;
+import java.util.Optional;
+
+public interface ValueObject {
+
+    Optional<Object> get(String name);
+
+    default Optional<Integer> getInt(String name) {
+        return get(name, Integer.class);
+    }
+
+    default Optional<Long> getLong(String name) {
+        return get(name, Long.class);
+    }
+
+    default Optional<Duration> getDuration(String name) {
+        return getString(name)
+            .map(TimeUtils::parse);
+    }
+
+    default Optional<Date> getDate(String name) {
+        return get(name, Date.class);
+    }
+
+    default Optional<Double> getDouble(String name) {
+        return get(name, Double.class);
+    }
+
+    default Optional<String> getString(String name) {
+        return get(name, String.class);
+    }
+
+    default Optional<Boolean> getBoolean(String name) {
+        return get(name, Boolean.class);
+    }
+
+    default <T> Optional<T> get(String name, Class<T> type) {
+        return get(name)
+            .map(obj -> FastBeanCopier.DEFAULT_CONVERT.convert(obj, type, FastBeanCopier.EMPTY_CLASS_ARRAY));
+    }
+
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/micrometer/MeterRegistryManager.java

@@ -0,0 +1,41 @@
+package org.jetlinks.community.micrometer;
+
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import lombok.Setter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @author bsetfeng
+ * @author wanghzeng
+ * @since 1.0
+ **/
+@Component
+@Setter
+public class MeterRegistryManager {
+
+    private Map<String, MeterRegistry> meterRegistryMap = new HashMap<>();
+
+    @Autowired
+    private List<MeterRegistrySupplier> suppliers;
+
+    private MeterRegistry createMeterRegistry(String metric) {
+        return new CompositeMeterRegistry(Clock.SYSTEM,
+            suppliers.stream()
+                .map(supplier -> supplier.getMeterRegistry(metric))
+                .collect(Collectors.toList()));
+    }
+
+    public MeterRegistry getMeterRegister(String metric) {
+        return meterRegistryMap.computeIfAbsent(metric, this::createMeterRegistry);
+    }
+
+
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/micrometer/MeterRegistrySupplier.java

@@ -0,0 +1,10 @@
+package org.jetlinks.community.micrometer;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+public interface MeterRegistrySupplier {
+
+    MeterRegistry getMeterRegistry(String metric);
+
+
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/ErrorUtils.java

@@ -0,0 +1,21 @@
+package org.jetlinks.community.utils;
+
+import org.hswebframework.web.authorization.exception.AccessDenyException;
+import org.hswebframework.web.exception.NotFoundException;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author wangzheng
+ * @see
+ * @since 1.0
+ */
+public class ErrorUtils {
+
+    public static <T> Mono<T> notFound(String message){
+        return Mono.error(()->new NotFoundException(message));
+    }
+
+    public static <T> Mono<T> accessDeny(String message){
+        return Mono.error(()->new AccessDenyException(message));
+    }
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/TimeUtils.java

@@ -0,0 +1,53 @@
+package org.jetlinks.community.utils;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+
+public class TimeUtils {
+
+
+    /**
+     * 将时间字符解析为{@link Duration}.如: 1d, 15m, 1h15m.
+     * 支持天(D,d),时(H,h),分(M,m),秒(s),毫秒(S)
+     *
+     * @param timeString 时间字符串
+     * @return Duration
+     */
+    public static Duration parse(String timeString) {
+
+        char[] all = timeString.toCharArray();
+        if ((all[0] == 'P') || (all[0] == '-' && all[1] == 'P')) {
+            return Duration.parse(timeString);
+        }
+        Duration duration = Duration.ofSeconds(0);
+        char[] tmp = new char[32];
+        int numIndex = 0;
+        for (char c : all) {
+            if (c == '-' || (c >= '0' && c <= '9')) {
+                tmp[numIndex++] = c;
+                continue;
+            }
+            long val = new BigDecimal(tmp, 0, numIndex).longValue();
+            numIndex = 0;
+            Duration plus = null;
+            if (c == 'D' || c == 'd') {
+                plus = Duration.ofDays(val);
+            } else if (c == 'H' || c == 'h') {
+                plus = Duration.ofHours(val);
+            } else if (c == 'M' || c == 'm') {
+                plus = Duration.ofMinutes(val);
+            } else if (c == 's') {
+                plus = Duration.ofSeconds(val);
+            } else if (c == 'S') {
+                plus = Duration.ofMillis(val);
+            } else if (c == 'W' || c == 'w') {
+                plus = Duration.ofDays(val * 7);
+            }
+            if (plus != null) {
+                duration = duration.plus(plus);
+            }
+        }
+        return duration;
+    }
+
+}

jetlinks-components/dashboard-components/pom.xml

@@ -9,7 +9,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>dashboard-components</artifactId>
+    <artifactId>dashboard-component</artifactId>
 
     <dependencies>
         <dependency>
@@ -23,6 +23,13 @@
             <artifactId>hsweb-easy-orm-rdb</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>common-component</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+
     </dependencies>
 
 </project>

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/CommonDimensionDefinition.java

@@ -11,8 +11,9 @@ import lombok.Getter;
 @AllArgsConstructor
 @Getter
 public enum CommonDimensionDefinition implements DimensionDefinition {
-    realTime("实时"),
-    history("历史");
+    realTime("实时数据"),
+    history("历史数据"),
+    agg("聚合数据");
 
     private String name;
 

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/MeasurementParameter.java

@@ -0,0 +1,24 @@
+package org.jetlinks.community.dashboard;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.jetlinks.community.ValueObject;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+@Getter
+@Setter
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class MeasurementParameter implements ValueObject {
+    private Map<String, Object> params = new HashMap<>();
+
+    public Optional<Object> get(String name) {
+        return Optional.ofNullable(params.get(name));
+    }
+
+}

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/SimpleMeasurementValue.java

@@ -0,0 +1,28 @@
+package org.jetlinks.community.dashboard;
+
+import lombok.*;
+import org.hswebframework.utils.time.DateFormatter;
+
+import java.util.Date;
+
+@Getter
+@Setter
+@Builder
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class SimpleMeasurementValue implements MeasurementValue {
+
+    private Object value;
+
+    private String timeString;
+
+    private long timestamp;
+
+    public static SimpleMeasurementValue of(Object value, Date time) {
+        return of(value, DateFormatter.toString(time, "yyyy-MM-dd HH:mm:ss"), time.getTime());
+    }
+
+    public static SimpleMeasurementValue of(Object value, long time) {
+        return of(value, new Date(time));
+    }
+}

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/measurements/SystemCpuMeasurementProvider.java

@@ -2,13 +2,13 @@ package org.jetlinks.community.dashboard.measurements;
 
 import lombok.SneakyThrows;
 import org.hswebframework.utils.time.DateFormatter;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.unit.UnifyUnit;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 
@@ -29,6 +29,7 @@ import static java.math.BigDecimal.ROUND_HALF_UP;
  * <pre>
  *     /dashboard/systemMonitor/cpu/usage/realTime
  * </pre>
+ *
  * @author zhouhao
  */
 @Component
@@ -106,11 +107,16 @@ public class SystemCpuMeasurementProvider
             return processCpuUsage.call() * 100;
         }
 
+        @SneakyThrows
+        public double getSystemCpu() {
+            return systemCpuUsage.call() * 100;
+        }
+
         @Override
         public Flux<MeasurementValue> getValue(MeasurementParameter parameter) {
-            // TODO: 2020/1/15 性能优化
+            //每秒获取系统CPU使用率
             return Flux.interval(Duration.ofSeconds(1))
-                .map(t -> SimpleMeasurementValue.of(BigDecimal.valueOf(getProcessCpu()).setScale(1, ROUND_HALF_UP),
+                .map(t -> SimpleMeasurementValue.of(BigDecimal.valueOf(getSystemCpu()).setScale(1, ROUND_HALF_UP),
                     DateFormatter.toString(new Date(), "HH:mm:ss"),
                     System.currentTimeMillis()))
                 .cast(MeasurementValue.class);

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/measurements/SystemMemoryMeasurementProvider.java

@@ -3,19 +3,21 @@ package org.jetlinks.community.dashboard.measurements;
 import lombok.Getter;
 import lombok.Setter;
 import org.hswebframework.utils.time.DateFormatter;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.SimplePropertyMetadata;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.types.LongType;
 import org.jetlinks.core.metadata.types.ObjectType;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 
-import java.lang.management.*;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.Date;

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/supports/CompositeDashboard.java

@@ -7,8 +7,10 @@ import org.jetlinks.community.dashboard.DashboardObject;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 class CompositeDashboard implements Dashboard {
 
@@ -21,6 +23,8 @@ class CompositeDashboard implements Dashboard {
 
     private Map<String, DashboardObject> staticObjects = new ConcurrentHashMap<>();
 
+    private List<Dashboard> staticDashboard = new CopyOnWriteArrayList<>();
+
     public void addProvider(MeasurementProvider provider) {
 
         DashboardObject object = staticObjects.computeIfAbsent(provider.getObjectDefinition().getId(), __ -> new CompositeDashboardObject());
@@ -31,17 +35,26 @@ class CompositeDashboard implements Dashboard {
 
     }
 
+    public void addDashboard(Dashboard dashboard){
+        staticDashboard.add(dashboard);
+    }
+
     public void addObject(DashboardObject object) {
         staticObjects.put(object.getDefinition().getId(), object);
     }
 
     @Override
     public Flux<DashboardObject> getObjects() {
-        return Flux.fromIterable(staticObjects.values());
+        return Flux.concat(
+            Flux.fromIterable(staticObjects.values()),
+            Flux.fromIterable(staticDashboard).flatMap(Dashboard::getObjects));
     }
 
     @Override
     public Mono<DashboardObject> getObject(String id) {
-        return Mono.justOrEmpty(staticObjects.get(id));
+        return Mono.justOrEmpty(staticObjects.get(id))
+            .switchIfEmpty(Mono.defer(()-> Flux.fromIterable(staticDashboard)
+                .flatMap(dashboard -> dashboard.getObject(id))
+                .next()));
     }
 }

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/supports/DefaultDashboardManager.java

@@ -1,8 +1,8 @@
 package org.jetlinks.community.dashboard.supports;
 
+import org.jetlinks.community.dashboard.Dashboard;
 import org.jetlinks.community.dashboard.DashboardDefinition;
 import org.jetlinks.community.dashboard.DashboardManager;
-import org.jetlinks.community.dashboard.Dashboard;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.stereotype.Component;
@@ -15,7 +15,7 @@ import java.util.concurrent.ConcurrentHashMap;
 @Component
 public class DefaultDashboardManager implements DashboardManager, BeanPostProcessor {
 
-    private Map<String, Dashboard> dashboards = new ConcurrentHashMap<>();
+    private Map<String, CompositeDashboard> dashboards = new ConcurrentHashMap<>();
 
     @Override
     public Flux<Dashboard> getDashboards() {
@@ -32,18 +32,17 @@ public class DefaultDashboardManager implements DashboardManager, BeanPostProces
 
         DashboardDefinition definition = provider.getDashboardDefinition();
 
-        Dashboard dashboard = dashboards.computeIfAbsent(definition.getId(), __ -> new CompositeDashboard(definition));
+        CompositeDashboard dashboard = dashboards.computeIfAbsent(definition.getId(), __ -> new CompositeDashboard(definition));
 
-        if (dashboard instanceof CompositeDashboard) {
-            CompositeDashboard compose = ((CompositeDashboard) dashboard);
-            compose.addProvider(provider);
-        } else {
-            throw new UnsupportedOperationException("unsupported register dashboard object : " + provider);
-        }
+        dashboard.addProvider(provider);
     }
 
     private void addDashboard(Dashboard dashboard) {
-        dashboards.put(dashboard.getDefinition().getId(), dashboard);
+
+        CompositeDashboard cached = dashboards.computeIfAbsent(dashboard.getDefinition().getId(), __ -> new CompositeDashboard(dashboard.getDefinition()));
+
+        cached.addDashboard(dashboard);
+
     }
 
     @Override

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/supports/MeasurementProvider.java

@@ -1,7 +1,6 @@
 package org.jetlinks.community.dashboard.supports;
 
 import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.measurements.SystemObjectDefinition;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -15,7 +14,7 @@ public interface MeasurementProvider {
 
     /**
      * @return 对象定义
-     * @see SystemObjectDefinition
+     * @see org.jetlinks.community.dashboard.measurements.SystemObjectDefinition
      */
     ObjectDefinition getObjectDefinition();
 

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/web/DashboardController.java

@@ -2,14 +2,11 @@ package org.jetlinks.community.dashboard.web;
 
 import com.alibaba.fastjson.JSON;
 import org.hswebframework.web.exception.NotFoundException;
-import org.jetlinks.community.dashboard.DashboardManager;
-import org.jetlinks.community.dashboard.DashboardObject;
-import org.jetlinks.community.dashboard.MeasurementParameter;
-import org.jetlinks.community.dashboard.MeasurementValue;
-import org.jetlinks.community.dashboard.web.response.DashboardMeasurementResponse;
-import org.jetlinks.community.dashboard.web.response.MeasurementInfo;
+import org.jetlinks.community.dashboard.*;
 import org.jetlinks.community.dashboard.web.request.DashboardMeasurementRequest;
 import org.jetlinks.community.dashboard.web.response.DashboardInfo;
+import org.jetlinks.community.dashboard.web.response.DashboardMeasurementResponse;
+import org.jetlinks.community.dashboard.web.response.MeasurementInfo;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
@@ -59,6 +56,30 @@ public class DashboardController {
             .flatMapMany(dim -> dim.getValue(MeasurementParameter.of(params)));
     }
 
+    /**
+     * POST 方式批量获取仪表数据,不支持获取实时数据.
+     *
+     * @param requests 请求参数
+     * @return 仪表数据
+     */
+    @PostMapping(value = "/_multi")
+    public Flux<DashboardMeasurementResponse> getMultiMeasurementValue(@RequestBody Flux<DashboardMeasurementRequest> requests) {
+        return requests.flatMap(request -> dashboardManager
+            .getDashboard(request.getDashboard())
+            .flatMap(dash -> dash.getObject(request.getObject()))
+            .flatMap(obj -> obj.getMeasurement(request.getMeasurement()))
+            .flatMap(meas -> meas.getDimension(request.getDimension()))
+            .filter(dim -> !dim.isRealTime()) //实时数据请使用
+            .flatMapMany(dim -> dim.getValue(MeasurementParameter.of(request.getParams())))
+            .map(val -> DashboardMeasurementResponse.of(request.getGroup(), val)));
+    }
+
+    /**
+     * 使用EventSource方式批量获取仪表数据,支持获取实时数据.
+     *
+     * @param requestJson 请求集合json
+     * @return 仪表数据
+     */
     @GetMapping(value = "/_multi", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     public Flux<DashboardMeasurementResponse> getMultiMeasurementValue(@RequestParam String requestJson) {
         return Flux.fromIterable(JSON.parseArray(requestJson, DashboardMeasurementRequest.class))

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/web/request/DashboardMeasurementRequest.java

@@ -0,0 +1,58 @@
+package org.jetlinks.community.dashboard.web.request;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.dashboard.Dashboard;
+import org.jetlinks.community.dashboard.DashboardDefinition;
+import org.jetlinks.community.dashboard.DimensionDefinition;
+import org.jetlinks.community.dashboard.MeasurementDefinition;
+import org.jetlinks.community.dashboard.web.response.DashboardMeasurementResponse;
+
+import java.util.Map;
+
+/**
+ * 仪表盘指标数据请求
+ *
+ * @author zhouhao
+ * @since 1.0
+ */
+@Getter
+@Setter
+public class DashboardMeasurementRequest {
+
+    /**
+     * 分组
+     * @see DashboardMeasurementResponse#getGroup()
+     */
+    private String group;
+
+    /**
+     * 仪表盘,如: device
+     * @see Dashboard#getDefinition()
+     */
+    private String dashboard;
+
+    /**
+     * 仪表对象,如: device1
+     * @see  DashboardDefinition#getId()
+     */
+    private String object;
+
+    /**
+     * 指标,如: 属性ID
+     * @see  MeasurementDefinition#getId()
+     */
+    private String measurement;
+
+    /**
+     * 维度
+     * @see DimensionDefinition#getId()
+     */
+    private String dimension;
+
+    /**
+     * 查询参数
+     */
+    private Map<String, Object> params;
+
+}

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/MeasurementParameter.java

@@ -1,19 +0,0 @@
-package org.jetlinks.community.dashboard;
-
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-import org.hswebframework.ezorm.core.param.QueryParam;
-
-import java.util.HashMap;
-import java.util.Map;
-
-@Getter
-@Setter
-@AllArgsConstructor(staticName = "of")
-@NoArgsConstructor
-public class MeasurementParameter {
-    private Map<String,Object> params=new HashMap<>();
-
-}

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/SimpleMeasurementValue.java

@@ -1,18 +0,0 @@
-package org.jetlinks.community.dashboard;
-
-import lombok.*;
-
-@Getter
-@Setter
-@Builder
-@AllArgsConstructor(staticName = "of")
-@NoArgsConstructor
-public class SimpleMeasurementValue implements MeasurementValue {
-
-    private Object value;
-
-    private String timeString;
-
-    private long timestamp;
-
-}

jetlinks-components/dashboard-components/src/main/java/org/jetlinks/community/dashboard/web/request/DashboardMeasurementRequest.java

@@ -1,24 +0,0 @@
-package org.jetlinks.community.dashboard.web.request;
-
-import lombok.Getter;
-import lombok.Setter;
-
-import java.util.Map;
-
-@Getter
-@Setter
-public class DashboardMeasurementRequest {
-
-    private String group;
-
-    private String dashboard;
-
-    private String object;
-
-    private String measurement;
-
-    private String dimension;
-
-    private Map<String,Object> params;
-
-}

jetlinks-components/elasticsearch-component/pom.xml

@@ -17,10 +17,12 @@
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-core</artifactId>
         </dependency>
+
         <dependency>
             <groupId>org.hswebframework</groupId>
             <artifactId>hsweb-easy-orm-elasticsearch</artifactId>
         </dependency>
+
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
@@ -41,6 +43,12 @@
             <artifactId>jetlinks-core</artifactId>
             <version>${jetlinks.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>timeseries-component</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         
     </dependencies>
 

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/ElasticRestClient.java

@@ -3,8 +3,6 @@ package org.jetlinks.community.elastic.search;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
-import org.apache.http.HttpHost;
-import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestHighLevelClient;
 
 /**

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/DefaultAggregationService.java

@@ -53,44 +53,45 @@ public class DefaultAggregationService implements AggregationService {
                                                     MetricsAggregationStructure structure,
                                                     ElasticIndex provider) {
         return searchSourceBuilderMono(queryParam, provider)
-                .doOnNext(builder -> builder.aggregation(
-                        structure.getType().aggregationBuilder(structure.getName(), structure.getField())))
-                .map(builder -> new SearchRequest(provider.getStandardIndex())
-                        .source(builder))
-                .flatMap(request -> Mono.<SearchResponse>create(monoSink ->
-                        restClient.getQueryClient().searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(monoSink))))
-                .map(searchResponse -> structure.getType().getResponse(structure.getName(), searchResponse));
+            .doOnNext(builder -> builder.aggregation(
+                structure.getType().aggregationBuilder(structure.getName(), structure.getField())))
+            .map(builder -> new SearchRequest(provider.getStandardIndex())
+                .source(builder))
+            .flatMap(request -> Mono.<SearchResponse>create(monoSink ->
+                restClient.getQueryClient().searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(monoSink))))
+            .map(searchResponse -> structure.getType().getResponse(structure.getName(), searchResponse));
     }
 
     @Override
     public Mono<BucketResponse> bucketAggregation(QueryParam queryParam, BucketAggregationsStructure structure, ElasticIndex provider) {
         return searchSourceBuilderMono(queryParam, provider)
-                .doOnNext(builder ->
-                        builder.aggregation(structure.getType().aggregationBuilder(structure))
-                )
-                .map(builder -> new SearchRequest(provider.getStandardIndex())
-                        .source(builder))
-                .doOnNext(searchRequest ->
-                        log.debug("聚合查询index:{},参数:{}",
-                                provider.getStandardIndex(),
-                                JSON.toJSON(searchRequest.source().toString())))
-                .flatMap(request -> Mono.<SearchResponse>create(monoSink ->
-                        restClient.getQueryClient().searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(monoSink))))
-                .map(response -> structure.getType().convert(response.getAggregations().get(structure.getName())))
-                .map(buckets -> BucketResponse.builder()
-                        .name(structure.getName())
-                        .buckets(buckets)
-                        .build()
-                )
-                ;
+            .doOnNext(builder ->
+                builder.aggregation(structure.getType().aggregationBuilder(structure))
+            )
+            .map(builder -> new SearchRequest(provider.getStandardIndex())
+                .source(builder))
+            .doOnNext(searchRequest ->
+                log.debug("聚合查询index:{},参数:{}",
+                    provider.getStandardIndex(),
+                    JSON.toJSON(searchRequest.source().toString())))
+            .flatMap(request -> Mono.<SearchResponse>create(monoSink ->
+                restClient.getQueryClient().searchAsync(request, RequestOptions.DEFAULT, translatorActionListener(monoSink))))
+            .map(response -> structure.getType().convert(response.getAggregations().get(structure.getName())))
+            .map(buckets -> BucketResponse.builder()
+                .name(structure.getName())
+                .buckets(buckets)
+                .build()
+            )
+            ;
 
     }
 
     private Mono<SearchSourceBuilder> searchSourceBuilderMono(QueryParam queryParam, ElasticIndex provider) {
-        queryParam.setPaging(false);
+        QueryParam tempQueryParam = queryParam.clone();
+        tempQueryParam.setPaging(false);
         return indexOperationService.getIndexMappingMetadata(provider.getStandardIndex())
-                .map(metadata -> translateService.translate(queryParam, metadata))
-                .doOnError(e -> log.error("解析queryParam错误, index:{}", provider.getStandardIndex(), e));
+            .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())));
     }
 

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/bucket/AggregationResponseHandle.java

@@ -24,58 +24,60 @@ public class AggregationResponseHandle {
     public static <A extends Aggregation> List<Bucket> terms(A a) {
         Terms terms = (Terms) a;
         return terms.getBuckets()
-                .stream()
-                .map(b -> {
-                    Bucket bucket = Bucket.builder()
-                            .key(b.getKeyAsString())
-                            .count(b.getDocCount())
-                            .build();
-                    b.getAggregations().asList()
-                            .forEach(subAggregation -> route(bucket, subAggregation));
-                    return bucket;
-                }).collect(Collectors.toList())
-                ;
+            .stream()
+            .map(b -> {
+                Bucket bucket = Bucket.builder()
+                    .key(b.getKeyAsString())
+                    .count(b.getDocCount())
+                    .name(a.getName())
+                    .build();
+                b.getAggregations().asList()
+                    .forEach(subAggregation -> route(bucket, subAggregation));
+                return bucket;
+            }).collect(Collectors.toList())
+            ;
     }
 
     public static <A extends Aggregation> List<Bucket> range(A a) {
         Range range = (Range) a;
         return range.getBuckets()
-                .stream()
-                .map(b -> {
-                    Bucket bucket = Bucket.builder()
-                            .key(b.getKeyAsString())
-                            .from(b.getFrom())
-                            .to(b.getTo())
-                            .fromAsString(b.getFromAsString())
-                            .toAsString(b.getToAsString())
-                            .count(b.getDocCount()).build();
-                    b.getAggregations().asList()
-                            .forEach(subAggregation -> {
-                                route(bucket, subAggregation);
-                            });
-                    return bucket;
-                }).collect(Collectors.toList())
-                ;
+            .stream()
+            .map(b -> {
+                Bucket bucket = Bucket.builder()
+                    .key(b.getKeyAsString())
+                    .from(b.getFrom())
+                    .to(b.getTo())
+                    .fromAsString(b.getFromAsString())
+                    .toAsString(b.getToAsString())
+                    .count(b.getDocCount()).build();
+                b.getAggregations().asList()
+                    .forEach(subAggregation -> {
+                        route(bucket, subAggregation);
+                    });
+                return bucket;
+            }).collect(Collectors.toList())
+            ;
     }
 
     public static <A extends Aggregation> List<Bucket> dateHistogram(A a) {
         Histogram histogram = (Histogram) a;
-        return bucketsHandle(histogram.getBuckets());
+        return bucketsHandle(histogram.getBuckets(), a.getName());
     }
 
-    private static List<Bucket> bucketsHandle(List<? extends Histogram.Bucket> buckets) {
+    private static List<Bucket> bucketsHandle(List<? extends Histogram.Bucket> buckets, String name) {
         return buckets
-                .stream()
-                .map(b -> {
-                    Bucket bucket = Bucket.builder()
-                            .key(b.getKeyAsString())
-                            .count(b.getDocCount())
-                            .build();
-                    b.getAggregations().asList()
-                            .forEach(subAggregation -> route(bucket, subAggregation));
-                    return bucket;
-                }).collect(Collectors.toList())
-                ;
+            .stream()
+            .map(b -> {
+                Bucket bucket = Bucket.builder()
+                    .key(b.getKeyAsString())
+                    .count(b.getDocCount())
+                    .name(name)
+                    .build();
+                b.getAggregations().asList()
+                    .forEach(subAggregation -> route(bucket, subAggregation));
+                return bucket;
+            }).collect(Collectors.toList())
+            ;
     }
 
     private static <A extends Aggregation> void route(Bucket bucket, A a) {
@@ -103,52 +105,60 @@ public class AggregationResponseHandle {
     public static <A extends Aggregation> MetricsResponseSingleValue avg(A a) {
         Avg avg = (Avg) a;
         return MetricsResponseSingleValue.builder()
-                .value(avg.getValue())
-                .valueAsString(avg.getValueAsString())
-                .build();
+            .value(avg.getValue())
+            .name(a.getName())
+            .valueAsString(avg.getValueAsString())
+            .build();
     }
 
     public static <A extends Aggregation> MetricsResponseSingleValue max(A a) {
         Max max = (Max) a;
         return MetricsResponseSingleValue.builder()
-                .value(max.getValue())
-                .valueAsString(max.getValueAsString())
-                .build();
+            .value(max.getValue())
+            .name(a.getName())
+            .valueAsString(max.getValueAsString())
+            .build();
     }
 
     public static <A extends Aggregation> MetricsResponseSingleValue min(A a) {
         Min min = (Min) a;
         return MetricsResponseSingleValue.builder()
-                .value(min.getValue())
-                .valueAsString(min.getValueAsString())
-                .build();
+            .value(min.getValue())
+            .name(a.getName())
+            .valueAsString(min.getValueAsString())
+            .build();
     }
 
     public static <A extends Aggregation> MetricsResponseSingleValue sum(A a) {
         Sum sum = (Sum) a;
         return MetricsResponseSingleValue.builder()
-                .value(sum.getValue())
-                .valueAsString(sum.getValueAsString())
-                .build();
+            .value(sum.getValue())
+            .name(a.getName())
+            .valueAsString(sum.getValueAsString())
+            .build();
     }
 
     public static <A extends Aggregation> void stats(Bucket bucket, A a) {
         Stats stats = (Stats) a;
         bucket.setAvg(MetricsResponseSingleValue.builder()
-                .value(stats.getAvg())
-                .valueAsString(stats.getAvgAsString())
-                .build());
+            .value(stats.getAvg())
+            .name(a.getName())
+            .valueAsString(stats.getAvgAsString())
+            .build());
         bucket.setMax(MetricsResponseSingleValue.builder()
-                .value(stats.getMax())
-                .valueAsString(stats.getMaxAsString())
-                .build());
+            .value(stats.getMax())
+            .name(a.getName())
+            .valueAsString(stats.getMaxAsString())
+            .build());
         bucket.setMin(MetricsResponseSingleValue.builder()
-                .value(stats.getMin())
-                .valueAsString(stats.getMinAsString())
-                .build());
+            .value(stats.getMin())
+            .name(a.getName())
+            .valueAsString(stats.getMinAsString())
+            .build());
         bucket.setSum(MetricsResponseSingleValue.builder()
-                .value(stats.getSum())
-                .valueAsString(stats.getSumAsString())
-                .build());
+            .value(stats.getSum())
+            .name(a.getName())
+            .valueAsString(stats.getSumAsString())
+            .build());
     }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/bucket/Bucket.java

@@ -3,7 +3,9 @@ package org.jetlinks.community.elastic.search.aggreation.bucket;
 import lombok.*;
 import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsResponseSingleValue;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * @author bsetfeng
@@ -18,6 +20,8 @@ public class Bucket {
 
     private String key;
 
+    private String name;
+
     private long count;
 
     private String fromAsString;
@@ -39,4 +43,32 @@ public class Bucket {
     private MetricsResponseSingleValue max;
 
     private List<Bucket> buckets;
+
+    public Map<String, Number> toMap() {
+        Map<String, Number> map = new HashMap<>();
+        if (this.sum != null) {
+            map.put(sum.getName(), sum.getValue());
+        }
+        if (this.valueCount != null) {
+            map.put(valueCount.getName(), valueCount.getValue());
+        }
+        if (this.avg != null) {
+            map.put(avg.getName(), avg.getValue());
+        }
+        if (this.min != null) {
+            map.put(min.getName(), min.getValue());
+        }
+        if (this.max != null) {
+            map.put(max.getName(), max.getValue());
+        }
+//
+//        if (this.getBuckets() != null) {
+//            bucketFlatMap(this.getBuckets(), map);
+//        }
+        return map;
+    }
+
+//    private void bucketFlatMap(List<Bucket> buckets, Map<String, Number> map) {
+//        buckets.forEach(bucket -> map.putAll(bucket.toMap()));
+//    }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/bucket/BucketAggregationsStructure.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.elastic.search.aggreation.bucket;
 
 import lombok.*;
+import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
 import org.hswebframework.utils.StringUtils;
 import org.jetlinks.community.elastic.search.aggreation.enums.BucketType;
 import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsAggregationStructure;
@@ -36,6 +37,8 @@ public class BucketAggregationsStructure {
 
     private List<Ranges> ranges;
 
+    private ExtendedBounds extendedBounds;
+
     /**
      * 时间格式
      */

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/enums/BucketType.java

@@ -37,8 +37,8 @@ public enum BucketType {
         @Override
         public AggregationBuilder aggregationBuilder(BucketAggregationsStructure structure) {
             TermsAggregationBuilder builder = AggregationBuilders
-                    .terms(structure.getName())
-                    .field(structure.getField());
+                .terms(structure.getName())
+                .field(structure.getField());
             if (structure.getSize() != null) {
                 builder.size(structure.getSize());
             }
@@ -62,15 +62,15 @@ public enum BucketType {
         @Override
         public AggregationBuilder aggregationBuilder(BucketAggregationsStructure structure) {
             RangeAggregationBuilder builder = AggregationBuilders
-                    .range(structure.getName())
-                    .field(structure.getField());
+                .range(structure.getName())
+                .field(structure.getField());
             if (structure.getMissingValue() != null) {
                 builder.missing(structure.getMissingValue());
             }
             structure.getRanges()
-                    .forEach(ranges -> {
-                        builder.addRange(ranges.getKey(), (Double) ranges.getForm(), (Double) ranges.getTo());
-                    });
+                .forEach(ranges -> {
+                    builder.addRange(ranges.getKey(), (Double) ranges.getForm(), (Double) ranges.getTo());
+                });
             commonAggregationSetting(builder, structure);
             return builder;
         }
@@ -84,15 +84,15 @@ public enum BucketType {
         @Override
         public AggregationBuilder aggregationBuilder(BucketAggregationsStructure structure) {
             DateRangeAggregationBuilder builder = AggregationBuilders
-                    .dateRange(structure.getName())
-                    .field(structure.getField());
+                .dateRange(structure.getName())
+                .field(structure.getField());
             if (StringUtils.hasText(structure.getFormat())) {
                 builder.format(structure.getFormat());
             }
             structure.getRanges()
-                    .forEach(ranges -> {
-                        builder.addRange(ranges.getKey(), ranges.getForm().toString(), ranges.getTo().toString());
-                    });
+                .forEach(ranges -> {
+                    builder.addRange(ranges.getKey(), ranges.getForm().toString(), ranges.getTo().toString());
+                });
             if (structure.getMissingValue() != null) {
                 builder.missing(structure.getMissingValue());
             }
@@ -110,22 +110,30 @@ public enum BucketType {
         @Override
         public AggregationBuilder aggregationBuilder(BucketAggregationsStructure structure) {
             DateHistogramAggregationBuilder builder = AggregationBuilders
-                    .dateHistogram(structure.getName())
-                    .field(structure.getField());
+                .dateHistogram(structure.getName())
+                .field(structure.getField());
             if (StringUtils.hasText(structure.getFormat())) {
                 builder.format(structure.getFormat());
             }
             if (StringUtils.hasText(structure.getInterval())) {
                 builder.dateHistogramInterval(new DateHistogramInterval(structure.getInterval()));
             }
+            if (structure.getExtendedBounds() != null) {
+                builder.extendedBounds(structure.getExtendedBounds());
+            }
             if (structure.getMissingValue() != null) {
                 builder.missing(structure.getMissingValue());
             }
+            Sort sort = structure.getSort();
+            if (sort != null) {
+                builder.order(mapping.get(OrderBuilder.of(sort.getOrder(), sort.getType())));
+            }
             builder.timeZone(DateTimeZone.getDefault());
             commonAggregationSetting(builder, structure);
             return builder;
         }
 
+
         @Override
         public <A extends Aggregation> List<Bucket> convert(A a) {
             return AggregationResponseHandle.dateHistogram(a);
@@ -149,16 +157,16 @@ public enum BucketType {
 
     private static void addMetricsSubAggregation(AggregationBuilder builder, List<MetricsAggregationStructure> subMetricsAggregation) {
         subMetricsAggregation
-                .forEach(subStructure -> {
-                    builder.subAggregation(subStructure.getType().aggregationBuilder(subStructure.getName(), subStructure.getField()));
-                });
+            .forEach(subStructure -> {
+                builder.subAggregation(subStructure.getType().aggregationBuilder(subStructure.getName(), subStructure.getField()));
+            });
     }
 
     private static void addBucketSubAggregation(AggregationBuilder builder, List<BucketAggregationsStructure> subBucketAggregation) {
         subBucketAggregation
-                .forEach(subStructure -> {
-                    builder.subAggregation(subStructure.getType().aggregationBuilder(subStructure));
-                });
+            .forEach(subStructure -> {
+                builder.subAggregation(subStructure.getType().aggregationBuilder(subStructure));
+            });
     }
 
     @Getter
@@ -177,6 +185,10 @@ public enum BucketType {
         mapping.put(OrderBuilder.of("asc", OrderType.COUNT), BucketOrder.count(true));
         mapping.put(OrderBuilder.of("desc", OrderType.COUNT), BucketOrder.count(false));
         mapping.put(OrderBuilder.of("asc", OrderType.KEY), BucketOrder.key(true));
-        mapping.put(OrderBuilder.of("desc", OrderType.KEY), BucketOrder.count(false));
+        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());
     }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/enums/MetricsType.java

@@ -36,9 +36,9 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             Avg avg = response.getAggregations().get(name);
             return MetricsResponse.builder()
-                    .results(Collections.singletonMap(AVG,
-                            new MetricsResponseSingleValue(avg.getValue(), avg.getValueAsString())))
-                    .build();
+                .results(Collections.singletonMap(AVG,
+                    new MetricsResponseSingleValue(avg.getValue(), avg.getName(), avg.getValueAsString())))
+                .build();
         }
     },
     MAX("最大") {
@@ -51,12 +51,12 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             Max max = response.getAggregations().get(name);
             return MetricsResponse.builder()
-                    .results(Collections.singletonMap(MAX,
-                            new MetricsResponseSingleValue(max.getValue(), max.getValueAsString())))
-                    .build();
+                .results(Collections.singletonMap(MAX,
+                    new MetricsResponseSingleValue(max.getValue(), max.getName(), max.getValueAsString())))
+                .build();
         }
     },
-    VALUE_COUNT("非空值计数") {
+    COUNT("非空值计数") {
         @Override
         public AggregationBuilder aggregationBuilder(String name, String filed) {
             return AggregationBuilders.count(name).field(filed);
@@ -66,9 +66,9 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             ValueCount valueCount = response.getAggregations().get(name);
             return MetricsResponse.builder()
-                    .results(Collections.singletonMap(VALUE_COUNT,
-                            new MetricsResponseSingleValue(valueCount.getValue(), valueCount.getValueAsString())))
-                    .build();
+                .results(Collections.singletonMap(COUNT,
+                    new MetricsResponseSingleValue(valueCount.getValue(), valueCount.getName(), valueCount.getValueAsString())))
+                .build();
         }
     },
     MIN("最小") {
@@ -81,9 +81,9 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             Min min = response.getAggregations().get(name);
             return MetricsResponse.builder()
-                    .results(Collections.singletonMap(MIN,
-                            new MetricsResponseSingleValue(min.getValue(), min.getValueAsString())))
-                    .build();
+                .results(Collections.singletonMap(MIN,
+                    new MetricsResponseSingleValue(min.getValue(), min.getName(), min.getValueAsString())))
+                .build();
         }
     },
     SUM("总数") {
@@ -96,9 +96,9 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             Sum sum = response.getAggregations().get(name);
             return MetricsResponse.builder()
-                    .results(Collections.singletonMap(SUM,
-                            new MetricsResponseSingleValue(sum.getValue(), sum.getValueAsString())))
-                    .build();
+                .results(Collections.singletonMap(SUM,
+                    new MetricsResponseSingleValue(sum.getValue(), sum.getName(), sum.getValueAsString())))
+                .build();
         }
     },
     STATS("统计汇总") {
@@ -111,14 +111,14 @@ public enum MetricsType {
         public MetricsResponse getResponse(String name, SearchResponse response) {
             Stats stats = response.getAggregations().get(name);
             Map<MetricsType, MetricsResponseSingleValue> results = new HashMap<>();
-            results.put(AVG, new MetricsResponseSingleValue(stats.getAvg(), stats.getAvgAsString()));
-            results.put(MIN, new MetricsResponseSingleValue(stats.getMin(), stats.getMinAsString()));
-            results.put(MAX, new MetricsResponseSingleValue(stats.getMax(), stats.getMaxAsString()));
-            results.put(SUM, new MetricsResponseSingleValue(stats.getSum(), stats.getMaxAsString()));
-            results.put(VALUE_COUNT, new MetricsResponseSingleValue(stats.getCount(), String.valueOf(stats.getCount())));
+            results.put(AVG, new MetricsResponseSingleValue(stats.getAvg(), stats.getName(), stats.getAvgAsString()));
+            results.put(MIN, new MetricsResponseSingleValue(stats.getMin(), stats.getName(), stats.getMinAsString()));
+            results.put(MAX, new MetricsResponseSingleValue(stats.getMax(), stats.getName(), stats.getMaxAsString()));
+            results.put(SUM, new MetricsResponseSingleValue(stats.getSum(), stats.getName(), stats.getMaxAsString()));
+            results.put(COUNT, new MetricsResponseSingleValue(stats.getCount(), stats.getName(), String.valueOf(stats.getCount())));
             return MetricsResponse.builder()
-                    .results(results)
-                    .build();
+                .results(results)
+                .build();
         }
     };
 
@@ -129,4 +129,13 @@ public enum MetricsType {
 
     public abstract MetricsResponse getResponse(String name, SearchResponse response);
 
+    public static MetricsType of(String name) {
+        for (MetricsType type : MetricsType.values()) {
+            if (type.name().equalsIgnoreCase(name)) {
+                return type;
+            }
+        }
+        throw new UnsupportedOperationException("不支持的聚合度量类型：" + name);
+    }
+
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/metrics/MetricsAggregationStructure.java

@@ -21,7 +21,7 @@ public class MetricsAggregationStructure {
     private String name;
 
     @NonNull
-    private MetricsType type = MetricsType.VALUE_COUNT;
+    private MetricsType type = MetricsType.COUNT;
 
     /**
      * 缺失值

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/metrics/MetricsResponseSingleValue.java

@@ -18,6 +18,8 @@ public class MetricsResponseSingleValue {
 
     private String valueAsString;
 
+    private String name;
+
     public static MetricsResponseSingleValue empty() {
         return new MetricsResponseSingleValue();
     }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/enums/FieldDateFormat.java

@@ -5,6 +5,7 @@ import lombok.Getter;
 import org.hswebframework.web.dict.EnumDict;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * @author bsetfeng
@@ -28,10 +29,17 @@ public enum FieldDateFormat implements EnumDict<String> {
 
     private String text;
 
-    public static String getFormatStr(List<FieldDateFormat> dateFormats) {
+    public static String getFormat(List<FieldDateFormat> dateFormats) {
+        return getFormatStr(dateFormats.stream()
+            .map(FieldDateFormat::getValue)
+            .collect(Collectors.toList())
+        );
+    }
+
+    public static String getFormatStr(List<String> dateFormats) {
         StringBuffer format = new StringBuffer();
         for (int i = 0; i < dateFormats.size(); i++) {
-            format.append(dateFormats.get(i).getValue());
+            format.append(dateFormats.get(i));
             if (i != dateFormats.size() - 1) {
                 format.append("||");
             }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/enums/FieldType.java

@@ -7,30 +7,28 @@ import org.hswebframework.web.exception.NotFoundException;
 import org.springframework.util.StringUtils;
 
 @AllArgsConstructor
-
+@Getter
 public enum FieldType implements EnumDict<String> {
 
-    TEXT("text"),
-    BYTE("byte"),
-    SHORT("short"),
-    INTEGER("integer"),
-    LONG("long"),
-    DATE("date"),
-    HALF_FLOAT("half_float"),
-    FLOAT("float"),
-    DOUBLE("double"),
-    BOOLEAN("boolean"),
-    OBJECT("object"),
-    AUTO("auto"),
-    NESTED("nested"),
-    IP("ip"),
-    ATTACHMENT("attachment"),
-    KEYWORD("keyword");
+    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");
 
-    @Override
-    public String getText() {
-        return value;
-    }
+    @Getter
+    private String text;
 
     @Getter
     private String value;
@@ -45,4 +43,15 @@ public enum FieldType implements EnumDict<String> {
         }
         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 + "的枚举");
+    }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/enums/IndexPatternEnum.java

@@ -0,0 +1,18 @@
+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;
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/enums/IndexStrategyEnum.java

@@ -0,0 +1,17 @@
+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;
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/DefaultIndexOperationService.java

@@ -41,33 +41,38 @@ public class DefaultIndexOperationService implements IndexOperationService {
     @Override
     public Mono<Boolean> indexIsExists(String index) {
         return Mono.create(sink -> {
-            try {
-                GetIndexRequest request = new GetIndexRequest(index);
-                sink.success(restClient.getQueryClient().indices().exists(request, RequestOptions.DEFAULT));
-            } catch (Exception e) {
-                log.error("查询es index 是否存在失败", e);
-                sink.error(e);
-            }
+            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());
-                        }
+            .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 void onFailure(Exception e) {
+                        sink.error(e);
+                    }
+                });
+            }));
 
     }
 
@@ -75,54 +80,57 @@ public class DefaultIndexOperationService implements IndexOperationService {
     @Override
     public Mono<IndexMappingMetadata> getIndexMappingMetadata(String index) {
         return indicesMappingCenter.getIndexMappingMetaData(index)
-                .map(Mono::just).orElseGet(() ->
-                        getIndexMapping(index)
-                                .doOnNext(indicesMappingCenter::register));
+            .map(Mono::just).orElseGet(() ->
+                getIndexMapping(index)
+                    .doOnNext(indicesMappingCenter::register));
     }
 
     private Mono<IndexMappingMetadata> getIndexMapping(String index) {
         return indexIsExists(index)
-                .filter(Boolean::booleanValue)
-                .flatMap(bool -> Mono.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")));
-                            }
-
-                            @Override
-                            public void onFailure(Exception e) {
-                                sink.error(e);
-                            }
-                        });
-                    }
-                }));
+            .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);
+            .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;
                     }
-                    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());
-                    }
-                });
+                    indexMappingMetaData.setMetadata(SingleMappingMetadata.builder()
+                        .name(key)
+                        .type(FieldType.of(tempValue.get("type")))
+                        .build());
+                }
+            });
         return indexMappingMetaData;
     }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/IndexTemplateProvider.java

@@ -0,0 +1,12 @@
+package org.jetlinks.community.elastic.search.index;
+
+/**
+ * @author bsetfeng
+ * @since 1.0
+ **/
+public interface IndexTemplateProvider {
+
+    static String getIndexTemplate(String index) {
+        return index.concat("_template");
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/alias/DefaultIndexAliasOperationService.java

@@ -0,0 +1,66 @@
+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);
+                }
+            });
+        });
+
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/mapping/MappingFactory.java

@@ -1,8 +1,8 @@
 package org.jetlinks.community.elastic.search.index.mapping;
 
 import org.hswebframework.web.exception.BusinessException;
-import org.jetlinks.community.elastic.search.enums.FieldType;
 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;
@@ -44,7 +44,13 @@ public class MappingFactory {
 
     public MappingFactory addFieldDateFormat(FieldDateFormat... dateFormats) {
         continuityOperateHandle(flag);
-        filedMap.put("format", FieldDateFormat.getFormatStr(Arrays.asList(dateFormats)));
+        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;
     }
 

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/mapping/SingleMappingMetadata.java

@@ -1,8 +1,8 @@
 package org.jetlinks.community.elastic.search.index.mapping;
 
 import lombok.*;
-import org.jetlinks.community.elastic.search.enums.FieldType;
 import org.jetlinks.community.elastic.search.enums.FieldDateFormat;
+import org.jetlinks.community.elastic.search.enums.FieldType;
 
 /**
  * @author bsetfeng

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/template/DefaultIndexTemplateOperationService.java

@@ -0,0 +1,67 @@
+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);
+                    }
+                });
+            }));
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/DefaultIndexStrategyProvider.java

@@ -0,0 +1,50 @@
+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;
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/IndexManager.java

@@ -0,0 +1,20 @@
+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());
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/IndexPatternManager.java

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

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/IndexStrategyManager.java

@@ -0,0 +1,12 @@
+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);
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/IndexStrategyProvider.java

@@ -0,0 +1,37 @@
+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();
+
+
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/StandardsIndexInit.java

@@ -0,0 +1,21 @@
+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;
+    }
+
+
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/StandardsIndexManager.java

@@ -0,0 +1,21 @@
+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);
+
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/StandardsIndexManagerCenter.java

@@ -0,0 +1,124 @@
+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();
+                    }
+                }));
+        });
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/entity/IndexStrategy.java

@@ -0,0 +1,19 @@
+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;
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/strategy/IndexDayPattern.java

@@ -0,0 +1,26 @@
+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));
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/strategy/IndexMonthPattern.java

@@ -0,0 +1,27 @@
+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));
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/manager/strategy/IndexSuffixStrategy.java

@@ -0,0 +1,24 @@
+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);
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/parser/DefaultQueryParamTranslateService.java

@@ -34,7 +34,7 @@ public class DefaultQueryParamTranslateService extends AbstractQueryParamTransla
     }
 
     @Override
-    protected QueryBuilder  queryBuilder(QueryParam queryParam, IndexMappingMetadata mappingMetaData) {
+    protected QueryBuilder queryBuilder(QueryParam queryParam, IndexMappingMetadata mappingMetaData) {
         BoolQueryBuilder queryBuilders = QueryBuilders.boolQuery();
         queryParam.getTerms()
                 .forEach(term -> parser.process(term, t ->

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/DefaultElasticSearchService.java

@@ -19,12 +19,12 @@ import org.elasticsearch.client.core.CountResponse;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.jetlinks.community.elastic.search.ElasticRestClient;
-import org.jetlinks.community.elastic.search.index.mapping.IndexMappingMetadata;
 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.springframework.beans.factory.annotation.Autowired;
+import org.reactivestreams.Publisher;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import reactor.core.publisher.Flux;
@@ -32,13 +32,9 @@ import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 
-import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.time.Duration;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -58,13 +54,13 @@ public class DefaultElasticSearchService implements ElasticSearchService {
 
     FluxSink<Buffer> sink;
 
-    @Autowired
     public DefaultElasticSearchService(ElasticRestClient restClient,
                                        QueryParamTranslateService translateService,
                                        IndexOperationService indexOperationService) {
         this.restClient = restClient;
         this.translateService = translateService;
         this.indexOperationService = indexOperationService;
+        init();
     }
 
 
@@ -105,12 +101,19 @@ public class DefaultElasticSearchService implements ElasticSearchService {
         });
     }
 
+    @Override
+    public <T> Mono<Void> commit(ElasticIndex index, Publisher<T> data) {
+        return Flux.from(data)
+            .flatMap(d -> commit(index, d))
+            .then();
+    }
+
     @PreDestroy
     public void shutdown() {
         sink.complete();
     }
 
-    @PostConstruct
+    //@PostConstruct
     public void init() {
 
         FluxUtils.bufferRate(Flux.<Buffer>create(sink -> this.sink = sink),
@@ -243,14 +246,17 @@ public class DefaultElasticSearchService implements ElasticSearchService {
                 }
                 return request;
             })
+            .doOnError(e -> log.error("查询index:" + provider.getStandardIndex() + "元数据错误", e))
             .doOnNext(searchRequest -> log.debug("查询index：{},es查询参数:{}", provider.getStandardIndex(), searchRequest.source().toString()));
     }
 
     private Mono<CountRequest> countRequestStructure(QueryParam queryParam, ElasticIndex provider) {
-        queryParam.setPaging(false);
+        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(queryParam, metadata)))
+                .source(translateService.translate(tempQueryParam, metadata)))
             .doOnNext(searchRequest -> log.debug("查询index：{},es查询参数:{}", provider.getStandardIndex(), searchRequest.source().toString()));
     }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/ElasticSearchService.java

@@ -3,6 +3,7 @@ package org.jetlinks.community.elastic.search.service;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.jetlinks.community.elastic.search.index.ElasticIndex;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -21,4 +22,5 @@ public interface ElasticSearchService {
 
     <T> Mono<Void> commit(ElasticIndex index, Collection<T> payload);
 
+    <T> Mono<Void> commit(ElasticIndex index, Publisher<T> data);
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/IndexAliasOperationService.java

@@ -0,0 +1,15 @@
+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);
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/IndexTemplateOperationService.java

@@ -0,0 +1,15 @@
+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);
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/AbstractTimeSeriesService.java

@@ -0,0 +1,20 @@
+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;
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/ESAbstractTimeSeriesManager.java

@@ -0,0 +1,67 @@
+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<>();
+
+    private StandardsIndexManager standardsIndexManager;
+
+    private IndexOperationService indexOperationService;
+
+    private IndexTemplateOperationService indexTemplateOperationService;
+
+    // TODO: 2020/2/11 策略变更动态更新实现
+    protected LocalTimeSeriesMetric getLocalTimeSeriesMetric(TimeSeriesMetric metric) {
+        return localMetricMap.computeIfAbsent(metric.getId(), 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;
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/ESAggregationData.java

@@ -0,0 +1,24 @@
+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;
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/ESTimeSeriesManager.java

@@ -0,0 +1,122 @@
+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;
+
+    private final IndexOperationService indexOperationService;
+
+    private final IndexTemplateOperationService indexTemplateOperationService;
+
+    private final StandardsIndexManager standardsIndexManager;
+
+    public ESTimeSeriesManager(TimeSeriesServiceRegisterCenter timeSeriesServiceRegisterCenter,
+                               IndexOperationService indexOperationService,
+                               StandardsIndexManager standardsIndexManager,
+                               IndexTemplateOperationService indexTemplateOperationService) {
+        this.timeSeriesServiceRegisterCenter = timeSeriesServiceRegisterCenter;
+        this.indexOperationService = indexOperationService;
+        this.standardsIndexManager = standardsIndexManager;
+        this.indexTemplateOperationService = indexTemplateOperationService;
+    }
+
+    @Override
+    public StandardsIndexManager getStandardsIndexManager() {
+        return this.standardsIndexManager;
+    }
+
+    @Override
+    public IndexOperationService getIndexOperationService() {
+        return indexOperationService;
+    }
+
+    @Override
+    public IndexTemplateOperationService getIndexTemplateOperationService() {
+        return indexTemplateOperationService;
+    }
+
+    @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();
+        }
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/ESTimeSeriesService.java

@@ -0,0 +1,239 @@
+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()
+        ).flatMapMany(bucketResponse -> Flux.fromIterable(new BucketsParser(bucketResponse.getBuckets()).result)
+            .take(param.getLimit())
+            .map(ESAggregationData::new));
+    }
+
+    @Override
+    public Mono<Void> save(Publisher<TimeSeriesData> data) {
+        return elasticSearchService.commit(getDefaultIndex(), Flux.from(data)
+            .map(timeSeriesData -> {
+                Map<String, Object> map = timeSeriesData.getData();
+                map.put("timestamp", timeSeriesData.getTimestamp());
+                return map;
+            }));
+    }
+
+    @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());
+        }
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/IndexAliasProvider.java

@@ -0,0 +1,14 @@
+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"));
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/TimeSeriesServiceRegisterCenter.java

@@ -0,0 +1,32 @@
+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));
+    }
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/translate/QueryParamTranslator.java

@@ -4,11 +4,8 @@ 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.elasticsearch.search.builder.SearchSourceBuilder;
-import org.elasticsearch.search.sort.SortOrder;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.jetlinks.community.elastic.search.enums.LinkTypeEnum;
-import org.springframework.util.StringUtils;
 
 import java.util.Objects;
 
@@ -29,19 +26,19 @@ public class QueryParamTranslator {
         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));
-    }
+//    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));
+//    }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/utils/DateTimeUtils.java

@@ -6,7 +6,8 @@ import lombok.extern.slf4j.Slf4j;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * @author bsetfeng

jetlinks-components/gateway-component/pom.xml

@@ -23,6 +23,18 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>dashboard-component</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>timeseries-component</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
     </dependencies>
 
 </project>

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/DeviceGateway.java

@@ -60,7 +60,7 @@ public interface DeviceGateway {
      */
     Mono<Void> shutdown();
 
-   default boolean isAlive(){
-       return true;
-   }
+    default boolean isAlive() {
+        return true;
+    }
 }

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/annotation/Subscribe.java

@@ -0,0 +1,28 @@
+package org.jetlinks.community.gateway.annotation;
+
+import org.springframework.core.annotation.AliasFor;
+
+import java.lang.annotation.*;
+
+/**
+ * 订阅来自消息网关的消息
+ *
+ * @author zhouhao
+ * @see org.jetlinks.community.gateway.MessageSubscriber
+ * @since 1.0
+ */
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Inherited
+@Documented
+public @interface Subscribe {
+
+    @AliasFor("value")
+    String[] topics() default {};
+
+    @AliasFor("topics")
+    String[] value() default {};
+
+    boolean shareCluster() default false;
+
+}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java

@@ -0,0 +1,57 @@
+package org.jetlinks.community.gateway.monitor;
+
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Consumer;
+
+class CompositeDeviceGatewayMonitor implements DeviceGatewayMonitor {
+
+    private List<DeviceGatewayMonitor> monitors = new ArrayList<>();
+
+    public CompositeDeviceGatewayMonitor add(DeviceGatewayMonitor... monitors) {
+        return add(Arrays.asList(monitors));
+    }
+
+    public CompositeDeviceGatewayMonitor add(Collection<DeviceGatewayMonitor> monitors) {
+        this.monitors.addAll(monitors);
+        return this;
+    }
+
+    protected void doWith(Consumer<DeviceGatewayMonitor> monitorConsumer) {
+        monitors.forEach(monitorConsumer);
+    }
+
+
+    @Override
+    public void totalConnection(long total) {
+        doWith(monitor -> monitor.totalConnection(total));
+    }
+
+    @Override
+    public void connected() {
+        doWith(DeviceGatewayMonitor::connected);
+    }
+
+    @Override
+    public void rejected() {
+        doWith(DeviceGatewayMonitor::rejected);
+    }
+
+    @Override
+    public void disconnected() {
+        doWith(DeviceGatewayMonitor::disconnected);
+    }
+
+    @Override
+    public void receivedMessage() {
+        doWith(DeviceGatewayMonitor::receivedMessage);
+    }
+
+    @Override
+    public void sentMessage() {
+        doWith(DeviceGatewayMonitor::sentMessage);
+    }
+}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeMessageGatewayMonitor.java

@@ -0,0 +1,66 @@
+package org.jetlinks.community.gateway.monitor;
+
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Consumer;
+
+class CompositeMessageGatewayMonitor implements MessageGatewayMonitor {
+
+   private List<MessageGatewayMonitor> monitors=new ArrayList<>();
+
+   public CompositeMessageGatewayMonitor add(MessageGatewayMonitor... monitors) {
+       return add(Arrays.asList(monitors));
+   }
+
+   public CompositeMessageGatewayMonitor add(Collection<MessageGatewayMonitor> monitors) {
+       this.monitors.addAll(monitors);
+       return this;
+   }
+
+   protected void doWith(Consumer<MessageGatewayMonitor> monitorConsumer) {
+       monitors.forEach(monitorConsumer);
+   }
+
+   @Override
+   public void totalSession(long sessionNumber) {
+       doWith(monitor->monitor.totalSession(sessionNumber));
+   }
+
+   @Override
+   public void acceptedSession() {
+       doWith(MessageGatewayMonitor::acceptedSession);
+   }
+
+   @Override
+   public void closedSession() {
+       doWith(MessageGatewayMonitor::closedSession);
+   }
+
+   @Override
+   public void subscribed() {
+       doWith(MessageGatewayMonitor::subscribed);
+   }
+
+   @Override
+   public void unsubscribed() {
+       doWith(MessageGatewayMonitor::unsubscribed);
+   }
+
+   @Override
+   public void dispatched(String connector) {
+       doWith(monitor->monitor.dispatched(connector));
+   }
+
+   @Override
+   public void acceptMessage() {
+       doWith(MessageGatewayMonitor::acceptMessage);
+   }
+
+   @Override
+   public void dispatchError(String connector, String sessionId, Throwable error) {
+       doWith(monitor->monitor.dispatchError(connector, sessionId, error));
+   }
+}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java

@@ -0,0 +1,40 @@
+package org.jetlinks.community.gateway.monitor;
+
+/**
+ * 设备网关监控
+ */
+public interface DeviceGatewayMonitor {
+
+    /**
+     * 上报总连接数
+     *
+     * @param total 总连接数
+     */
+    void totalConnection(long total);
+
+    /**
+     * 创建新连接
+     */
+    void connected();
+
+    /**
+     * 拒绝连接
+     */
+    void rejected();
+
+    /**
+     * 断开连接
+     */
+    void disconnected();
+
+    /**
+     * 接收消息
+     */
+    void receivedMessage();
+
+    /**
+     * 发送消息
+     */
+    void sentMessage();
+
+}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java

@@ -0,0 +1,6 @@
+package org.jetlinks.community.gateway.monitor;
+
+public interface DeviceGatewayMonitorSupplier {
+      DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags);
+
+}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java

@@ -0,0 +1,73 @@
+package org.jetlinks.community.gateway.monitor;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
+
+public class GatewayMonitors {
+
+    private static final List<MessageGatewayMonitorSupplier> suppliers = new CopyOnWriteArrayList<>();
+
+    private static final List<DeviceGatewayMonitorSupplier> deviceGatewayMonitorSuppliers = new CopyOnWriteArrayList<>();
+
+    static final NoneMessageGatewayMonitor none = new NoneMessageGatewayMonitor();
+
+    static final NoneDeviceGatewayMonitor nonDevice = new NoneDeviceGatewayMonitor();
+
+
+    static {
+
+    }
+
+    public static void register(MessageGatewayMonitorSupplier supplier) {
+        suppliers.add(supplier);
+    }
+
+
+    public static void register(DeviceGatewayMonitorSupplier supplier) {
+        deviceGatewayMonitorSuppliers.add(supplier);
+    }
+
+    private static MessageGatewayMonitor doGetMessageGatewayMonitor(String id, String... tags) {
+        List<MessageGatewayMonitor> all = suppliers.stream()
+            .map(supplier -> supplier.getMessageGatewayMonitor(id, tags))
+            .filter(Objects::nonNull)
+            .collect(Collectors.toList());
+
+        if (all.isEmpty()) {
+            return none;
+        }
+        if (all.size() == 1) {
+            return all.get(0);
+        }
+        CompositeMessageGatewayMonitor monitor = new CompositeMessageGatewayMonitor();
+        monitor.add(all);
+        return monitor;
+    }
+
+    private static DeviceGatewayMonitor doGetDeviceGatewayMonitor(String id, String... tags) {
+        List<DeviceGatewayMonitor> all = deviceGatewayMonitorSuppliers.stream()
+            .map(supplier -> supplier.getDeviceGatewayMonitor(id, tags))
+            .filter(Objects::nonNull)
+            .collect(Collectors.toList());
+
+        if (all.isEmpty()) {
+            return nonDevice;
+        }
+        if (all.size() == 1) {
+            return all.get(0);
+        }
+        CompositeDeviceGatewayMonitor monitor = new CompositeDeviceGatewayMonitor();
+        monitor.add(all);
+        return monitor;
+    }
+
+    public static MessageGatewayMonitor getMessageGatewayMonitor(String id, String... tags) {
+        return new LazyMessageGatewayMonitor(() -> doGetMessageGatewayMonitor(id, tags));
+    }
+
+    public static DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags) {
+        return new LazyDeviceGatewayMonitor(() -> doGetDeviceGatewayMonitor(id, tags));
+    }
+}