自动创建ES索引 & 主动填补指标历史曲线缺少的点

Author: zengqiao

km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/MetricESSender.java

@@ -5,7 +5,6 @@
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.*;
 import com.xiaojukeji.know.streaming.km.common.bean.po.BaseESPO;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.*;
-import com.xiaojukeji.know.streaming.km.common.enums.metric.KafkaMetricIndexEnum;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.EnvUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.NamedThreadFactory;
@@ -21,6 +20,8 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.*;
+
 @Component
 public class MetricESSender implements ApplicationListener<BaseMetricEvent> {
     protected static final ILog  LOGGER = LogFactory.getLog("METRIC_LOGGER");
@@ -41,37 +42,37 @@ public void init(){
     public void onApplicationEvent(BaseMetricEvent event) {
         if(event instanceof BrokerMetricEvent) {
             BrokerMetricEvent brokerMetricEvent = (BrokerMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.BROKER_INFO,
+            send2es(BROKER_INDEX,
                     ConvertUtil.list2List(brokerMetricEvent.getBrokerMetrics(), BrokerMetricPO.class)
             );
 
         } else if(event instanceof ClusterMetricEvent) {
             ClusterMetricEvent clusterMetricEvent = (ClusterMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.CLUSTER_INFO,
+            send2es(CLUSTER_INDEX,
                     ConvertUtil.list2List(clusterMetricEvent.getClusterMetrics(), ClusterMetricPO.class)
             );
 
         } else if(event instanceof TopicMetricEvent) {
             TopicMetricEvent topicMetricEvent = (TopicMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.TOPIC_INFO,
+            send2es(TOPIC_INDEX,
                     ConvertUtil.list2List(topicMetricEvent.getTopicMetrics(), TopicMetricPO.class)
             );
 
         } else if(event instanceof PartitionMetricEvent) {
             PartitionMetricEvent partitionMetricEvent = (PartitionMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.PARTITION_INFO,
+            send2es(PARTITION_INDEX,
                     ConvertUtil.list2List(partitionMetricEvent.getPartitionMetrics(), PartitionMetricPO.class)
             );
 
         } else if(event instanceof GroupMetricEvent) {
             GroupMetricEvent groupMetricEvent = (GroupMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.GROUP_INFO,
+            send2es(GROUP_INDEX,
                     ConvertUtil.list2List(groupMetricEvent.getGroupMetrics(), GroupMetricPO.class)
             );
 
         } else if(event instanceof ReplicaMetricEvent) {
             ReplicaMetricEvent replicaMetricEvent = (ReplicaMetricEvent)event;
-            send2es(KafkaMetricIndexEnum.REPLICATION_INFO,
+            send2es(REPLICATION_INDEX,
                     ConvertUtil.list2List(replicaMetricEvent.getReplicationMetrics(), ReplicationMetricPO.class)
             );
         }
@@ -80,19 +81,19 @@ public void onApplicationEvent(BaseMetricEvent event) {
     /**
      * 根据不同监控维度来发送
      */
-    private boolean send2es(KafkaMetricIndexEnum stats, List<? extends BaseESPO> statsList){
+    private boolean send2es(String index, List<? extends BaseESPO> statsList){
         if (CollectionUtils.isEmpty(statsList)) {
             return true;
         }
 
         if (!EnvUtil.isOnline()) {
             LOGGER.info("class=MetricESSender||method=send2es||ariusStats={}||size={}",
-                    stats.getIndex(), statsList.size());
+                    index, statsList.size());
         }
 
-        BaseMetricESDAO baseMetricESDao = BaseMetricESDAO.getByStatsType(stats);
+        BaseMetricESDAO baseMetricESDao = BaseMetricESDAO.getByStatsType(index);
         if (Objects.isNull( baseMetricESDao )) {
-            LOGGER.error("class=MetricESSender||method=send2es||errMsg=fail to find {}", stats.getIndex());
+            LOGGER.error("class=MetricESSender||method=send2es||errMsg=fail to find {}", index);
             return false;
         }
 

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/vo/metrics/point/MetricPointVO.java

@@ -29,6 +29,10 @@ public class MetricPointVO implements Comparable<MetricPointVO> {
     @Override
     public int compareTo(MetricPointVO o) {
         if(null == o){return 0;}
+        if(null == this.getTimeStamp()
+                || null == o.getTimeStamp()){
+            return 0;
+        }
 
         return this.getTimeStamp().intValue() - o.getTimeStamp().intValue();
     }