[Optimize]优化日志输出 & 本地缓存统一管理(#800)

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/cluster/impl/ClusterTopicsManagerImpl.java

@@ -44,7 +44,7 @@ public class ClusterTopicsManagerImpl implements ClusterTopicsManager {
         List<Topic> topicList = topicService.listTopicsFromDB(clusterPhyId);
 
         // 获取集群所有Topic的指标
-        Map<String, TopicMetrics> metricsMap = topicMetricService.getLatestMetricsFromCacheFirst(clusterPhyId);
+        Map<String, TopicMetrics> metricsMap = topicMetricService.getLatestMetricsFromCache(clusterPhyId);
 
         // 转换成vo
         List<ClusterPhyTopicsOverviewVO> voList = TopicVOConverter.convert2ClusterPhyTopicsOverviewVOList(topicList, metricsMap);

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/MsgConstant.java

@@ -52,6 +52,10 @@ public class MsgConstant {
 
     /**************************************************** Partition ****************************************************/
 
+    public static String getPartitionNoLeader(Long clusterPhyId) {
+        return String.format("集群ID:[%d] 所有分区NoLeader", clusterPhyId);
+    }
+
     public static String getPartitionNoLeader(Long clusterPhyId, String topicName) {
         return String.format("集群ID:[%d] Topic名称:[%s] 所有分区NoLeader", clusterPhyId, topicName);
     }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/DataBaseDataLocalCache.java

@@ -3,6 +3,7 @@ package com.xiaojukeji.know.streaming.km.core.cache;
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.ClusterMetrics;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.TopicMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.partition.Partition;
 
 import java.util.List;
@@ -10,6 +11,11 @@ import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 public class DataBaseDataLocalCache {
+    private static final Cache<Long, Map<String, TopicMetrics>> topicLatestMetricsCache = Caffeine.newBuilder()
+            .expireAfterWrite(360, TimeUnit.SECONDS)
+            .maximumSize(500)
+            .build();
+
     private static final Cache<Long, ClusterMetrics> clusterLatestMetricsCache = Caffeine.newBuilder()
             .expireAfterWrite(180, TimeUnit.SECONDS)
             .maximumSize(500)
@@ -20,6 +26,14 @@ public class DataBaseDataLocalCache {
             .maximumSize(500)
             .build();
 
+    public static Map<String, TopicMetrics> getTopicMetrics(Long clusterPhyId) {
+        return topicLatestMetricsCache.getIfPresent(clusterPhyId);
+    }
+
+    public static void putTopicMetrics(Long clusterPhyId, Map<String, TopicMetrics> metricsMap) {
+        topicLatestMetricsCache.put(clusterPhyId, metricsMap);
+    }
+
     public static ClusterMetrics getClusterLatestMetrics(Long clusterPhyId) {
         return clusterLatestMetricsCache.getIfPresent(clusterPhyId);
     }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/DatabaseDataFlusher.java

@@ -4,13 +4,18 @@ import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.ClusterMetrics;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.TopicMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.partition.Partition;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
 import com.xiaojukeji.know.streaming.km.core.cache.DataBaseDataLocalCache;
 import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterMetricService;
 import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
 import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService;
+import com.xiaojukeji.know.streaming.km.core.service.topic.TopicMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
+import com.xiaojukeji.know.streaming.km.persistence.cache.LoadedClusterPhyCache;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
@@ -18,11 +23,19 @@ import org.springframework.stereotype.Service;
 import javax.annotation.PostConstruct;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 @Service
 public class DatabaseDataFlusher {
     private static final ILog LOGGER  = LogFactory.getLog(DatabaseDataFlusher.class);
 
+    @Autowired
+    private TopicService topicService;
+
+    @Autowired
+    private TopicMetricService topicMetricService;
+
     @Autowired
     private ClusterPhyService clusterPhyService;
 
@@ -37,6 +50,8 @@ public class DatabaseDataFlusher {
         this.flushPartitionsCache();
 
         this.flushClusterLatestMetricsCache();
+
+        this.flushTopicLatestMetricsCache();
     }
 
     @Scheduled(cron="0 0/1 * * * ?")
@@ -81,4 +96,27 @@ public class DatabaseDataFlusher {
             });
         }
     }
+
+    @Scheduled(cron = "0 0/1 * * * ?")
+    private void flushTopicLatestMetricsCache() {
+        for (ClusterPhy clusterPhy: LoadedClusterPhyCache.listAll().values()) {
+            FutureUtil.quickStartupFutureUtil.submitTask(() -> {
+                try {
+
+                    List<String> topicNameList = topicService.listTopicsFromCacheFirst(clusterPhy.getId()).stream().map(Topic::getTopicName).collect(Collectors.toList());
+
+                    List<TopicMetrics> metricsList = topicMetricService.listTopicLatestMetricsFromES(clusterPhy.getId(), topicNameList, Collections.emptyList());
+
+                    Map<String, TopicMetrics> metricsMap = metricsList
+                            .stream()
+                            .collect(Collectors.toMap(TopicMetrics::getTopic, Function.identity()));
+
+                    DataBaseDataLocalCache.putTopicMetrics(clusterPhy.getId(), metricsMap);
+
+                } catch (Exception e) {
+                    LOGGER.error("method=flushTopicLatestMetricsCache||clusterPhyId={}||errMsg=exception!",  clusterPhy.getId(), e);
+                }
+            });
+        }
+    }
 }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/config/PlatformClusterConfigService.java

@@ -17,4 +17,5 @@ public interface PlatformClusterConfigService {
 
     Map<String, PlatformClusterConfigPO> getByClusterAndGroupWithoutDefault(Long clusterPhyId, String group);
 
+    Map<Long, Map<String, PlatformClusterConfigPO>> listByGroup(String groupName);
 }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/config/impl/PlatformClusterConfigServiceImpl.java

@@ -12,6 +12,7 @@ import com.xiaojukeji.know.streaming.km.persistence.mysql.config.PlatformCluster
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -68,4 +69,20 @@ public class PlatformClusterConfigServiceImpl implements PlatformClusterConfigSe
 
         return configPOMap;
     }
+
+    @Override
+    public Map<Long, Map<String, PlatformClusterConfigPO>> listByGroup(String groupName) {
+        LambdaQueryWrapper<PlatformClusterConfigPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+        lambdaQueryWrapper.eq(PlatformClusterConfigPO::getValueGroup, groupName);
+
+        List<PlatformClusterConfigPO> poList = platformClusterConfigDAO.selectList(lambdaQueryWrapper);
+
+        Map<Long, Map<String, PlatformClusterConfigPO>> poMap = new HashMap<>();
+        poList.forEach(elem -> {
+            poMap.putIfAbsent(elem.getClusterId(), new HashMap<>());
+            poMap.get(elem.getClusterId()).put(elem.getValueName(), elem);
+        });
+
+        return poMap;
+    }
 }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/TopicMetricService.java

@@ -23,7 +23,7 @@ public interface TopicMetricService {
     /**
      * 优先从本地缓存获取metrics信息
      */
-    Map<String, TopicMetrics> getLatestMetricsFromCacheFirst(Long clusterPhyId);
+    Map<String, TopicMetrics> getLatestMetricsFromCache(Long clusterPhyId);
 
     /**
      * 获取Topic在具体Broker上最新的一个指标

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/TopicMetricServiceImpl.java

@@ -2,13 +2,10 @@ package com.xiaojukeji.know.streaming.km.core.service.topic.impl;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import com.google.common.collect.Table;
 import com.xiaojukeji.know.streaming.km.common.bean.dto.metrices.MetricDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.dto.metrices.MetricsTopicDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.PartitionMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.TopicMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.VersionItemParam;
@@ -30,25 +27,22 @@ import com.xiaojukeji.know.streaming.km.common.utils.BeanUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.cache.CollectedMetricsLocalCache;
+import com.xiaojukeji.know.streaming.km.core.cache.DataBaseDataLocalCache;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
 import com.xiaojukeji.know.streaming.km.core.service.health.state.HealthStateService;
 import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionMetricService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicMetricService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
 import com.xiaojukeji.know.streaming.km.core.service.version.BaseMetricService;
-import com.xiaojukeji.know.streaming.km.persistence.cache.LoadedClusterPhyCache;
 import com.xiaojukeji.know.streaming.km.persistence.es.dao.TopicMetricESDAO;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaJMXClient;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 
 import javax.management.InstanceNotFoundException;
 import javax.management.ObjectName;
 import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus.*;
@@ -58,8 +52,7 @@ import static com.xiaojukeji.know.streaming.km.core.service.version.metrics.kafk
  */
 @Service
 public class TopicMetricServiceImpl extends BaseMetricService implements TopicMetricService {
-
+    private static final ILog LOGGER = LogFactory.getLog(TopicMetricServiceImpl.class);
-    private static final ILog LOGGER = LogFactory.getLog( TopicMetricServiceImpl.class);
 
     public static final String TOPIC_METHOD_DO_NOTHING                                              = "doNothing";
     public static final String TOPIC_METHOD_GET_HEALTH_SCORE                                        = "getMetricHealthScore";
@@ -86,18 +79,6 @@ public class TopicMetricServiceImpl extends BaseMetricService implements TopicMe
     @Autowired
     private TopicMetricESDAO topicMetricESDAO;
 
-    private final Cache<Long, Map<String, TopicMetrics>> topicLatestMetricsCache = Caffeine.newBuilder()
-            .expireAfterWrite(5, TimeUnit.MINUTES)
-            .maximumSize(200)
-            .build();
-
-    @Scheduled(cron = "0 0/2 * * * ?")
-    private void flushClusterLatestMetricsCache() {
-        for (ClusterPhy clusterPhy: LoadedClusterPhyCache.listAll().values()) {
-            this.updateCacheAndGetMetrics(clusterPhy.getId());
-        }
-    }
-
     @Override
     protected VersionItemTypeEnum getVersionItemType() {
         return VersionItemTypeEnum.METRIC_TOPIC;
@@ -152,13 +133,13 @@ public class TopicMetricServiceImpl extends BaseMetricService implements TopicMe
     }
 
     @Override
-    public Map<String, TopicMetrics> getLatestMetricsFromCacheFirst(Long clusterPhyId) {
+    public Map<String, TopicMetrics> getLatestMetricsFromCache(Long clusterPhyId) {
-        Map<String, TopicMetrics> metricsMap = topicLatestMetricsCache.getIfPresent(clusterPhyId);
+        Map<String, TopicMetrics> metricsMap = DataBaseDataLocalCache.getTopicMetrics(clusterPhyId);
-        if (metricsMap != null) {
+        if (metricsMap == null) {
-            return metricsMap;
+            return new HashMap<>();
         }
 
-        return this.updateCacheAndGetMetrics(clusterPhyId);
+        return metricsMap;
     }
 
     @Override
@@ -308,19 +289,8 @@ public class TopicMetricServiceImpl extends BaseMetricService implements TopicMe
         return Result.buildSuc(count);
     }
 
+
     /**************************************************** private method ****************************************************/
-    private Map<String, TopicMetrics> updateCacheAndGetMetrics(Long clusterPhyId) {
-        List<String> topicNames = topicService.listTopicsFromDB(clusterPhyId)
-                .stream().map(Topic::getTopicName).collect(Collectors.toList());
-
-        List<TopicMetrics> metrics = listTopicLatestMetricsFromES(clusterPhyId, topicNames, Arrays.asList());
-
-        Map<String, TopicMetrics> metricsMap = metrics.stream()
-                .collect(Collectors.toMap(TopicMetrics::getTopic, Function.identity()));
-
-        topicLatestMetricsCache.put(clusterPhyId, metricsMap);
-        return metricsMap;
-    }
 
 
     private List<String> listTopNTopics(Long clusterId, int topN){

km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/ESOpClient.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.didiglobal.logi.elasticsearch.client.ESClient;
 import com.didiglobal.logi.elasticsearch.client.gateway.document.ESIndexRequest;
 import com.didiglobal.logi.elasticsearch.client.gateway.document.ESIndexResponse;
+import com.didiglobal.logi.elasticsearch.client.model.exception.ESIndexNotFoundException;
 import com.didiglobal.logi.elasticsearch.client.model.type.ESVersion;
 import com.didiglobal.logi.elasticsearch.client.request.batch.BatchNode;
 import com.didiglobal.logi.elasticsearch.client.request.batch.BatchType;
@@ -20,10 +21,11 @@ import com.didiglobal.logi.elasticsearch.client.response.indices.puttemplate.ESI
 import com.didiglobal.logi.elasticsearch.client.response.query.query.ESQueryResponse;
 import com.didiglobal.logi.elasticsearch.client.response.setting.template.TemplateConfig;
 import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
 import com.google.common.collect.Lists;
 import com.xiaojukeji.know.streaming.km.common.bean.po.BaseESPO;
-import com.xiaojukeji.know.streaming.km.common.utils.EnvUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.LoggerUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpStatus;
@@ -36,6 +38,7 @@ import javax.annotation.Nullable;
 import javax.annotation.PostConstruct;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -43,7 +46,7 @@ import java.util.stream.Collectors;
 
 @Component
 public class ESOpClient {
-    private static final ILog LOGGER   = LogFactory.getLog("ES_LOGGER");
+    private static final ILog LOGGER = LoggerUtil.getESLogger();
 
     /**
      * es 地址
@@ -90,40 +93,27 @@ public class ESOpClient {
             ESClient esClient = this.buildEsClient(esAddress, esPass, "", "");
             if (esClient != null) {
                 this.esClientPool.add(esClient);
-                LOGGER.info("class=ESOpClient||method=init||msg=add new es client {}", esAddress);
+                LOGGER.info("method=init||esAddress={}||msg=add new es client", esAddress);
             }
         }
     }
 
     /**
-     * 从更新es http 客户端连接池找那个获取
+     * 获取ES客户端
-     *
-     * @return
      */
     public ESClient getESClientFromPool() {
+        if (ValidateUtils.isEmptyList(esClientPool)) {
+            return null;
+        }
+
         return esClientPool.get((int)(System.currentTimeMillis() % clientCnt));
     }
 
-    /**
-     * 归还到es http 客户端连接池
-     * @param esClient
-     */
-    public void returnESClientToPool(ESClient esClient) {
-        // 已不需要进行归还，后续再删除该代码
-    }
-
     /**
      * 查询并获取第一个元素
-     *
-     * @param indexName
-     * @param queryDsl
-     * @param clzz
-     * @param <T>
-     * @return
      */
-    public <T> T performRequestAndTakeFirst(String indexName, String queryDsl, Class<T> clzz) {
+    public <T> T performRequestAndTakeFirst(String indexName, String queryDsl, Class<T> clazz) {
-        List<T> hits = performRequest(indexName, queryDsl, clzz);
+        List<T> hits = this.performRequest(indexName, queryDsl, clazz);
-
         if (CollectionUtils.isEmpty(hits)) {
             return null;
         }
@@ -133,31 +123,20 @@ public class ESOpClient {
 
     /**
      * 查询并获取第一个元素
-     *
-     * @param indexName
-     * @param queryDsl
-     * @param clazz
-     * @param <T>
-     * @return
      */
-    public <T> T performRequestAndTakeFirst(String routingValue, String indexName,
+    public <T> T performRequestAndTakeFirst(String routingValue, String indexName, String queryDsl, Class<T> clazz) {
-                                            String queryDsl, Class<T> clazz) {
+        List<T> hits = this.performRequestWithRouting(routingValue, indexName, queryDsl, clazz);
-        List<T> hits = performRequestWithRouting(routingValue, indexName, queryDsl, clazz);
+        if (CollectionUtils.isEmpty(hits)) {
-
+            return null;
-        if (CollectionUtils.isEmpty(hits)) {return null;}
+        }
 
         return hits.get(0);
     }
 
     /**
      * 执行查询
-     *
-     * @param indexName
-     * @param queryDsl
-     * @return
-     * @throws IOException
      */
-    public ESQueryResponse performRequest(String indexName,String queryDsl) {
+    public ESQueryResponse performRequest(String indexName, String queryDsl) {
         return doQuery(new ESQueryRequest().indices(indexName).source(queryDsl));
     }
 
@@ -170,7 +149,7 @@ public class ESOpClient {
         return func.apply(esQueryResponse);
     }
 
-    public  <T> List<T> performRequest(String indexName, String queryDsl, Class<T> clzz) {
+    public <T> List<T> performRequest(String indexName, String queryDsl, Class<T> clzz) {
         ESQueryResponse esQueryResponse = doQuery(
                 new ESQueryRequest().indices(indexName).source(queryDsl).clazz(clzz));
         if (esQueryResponse == null) {
@@ -210,8 +189,7 @@ public class ESOpClient {
         return hits;
     }
 
-    public <R> R performRequestWithRouting(String routingValue, String indexName,
+    public <R> R performRequestWithRouting(String routingValue, String indexName, String queryDsl, Function<ESQueryResponse, R> func, int tryTimes) {
-                                           String queryDsl, Function<ESQueryResponse, R> func, int tryTimes) {
         ESQueryResponse esQueryResponse;
         do {
             esQueryResponse = doQuery(new ESQueryRequest().routing(routingValue).indices(indexName).source(queryDsl));
@@ -222,16 +200,12 @@ public class ESOpClient {
 
     /**
      * 写入单条
-     *
-     * @param source
-     * @return
      */
     public boolean index(String indexName, String id, String source) {
-        ESClient esClient = null;
         ESIndexResponse response = null;
 
         try {
-            esClient = getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
             if (esClient == null) {
                 return false;
             }
@@ -250,20 +224,11 @@ public class ESOpClient {
                 return response.getRestStatus().getStatus() == HttpStatus.SC_OK
                         || response.getRestStatus().getStatus() == HttpStatus.SC_CREATED;
             }
-
         } catch (Exception e) {
-            LOGGER.warn(
+            LOGGER.error(
-                    "class=ESOpClient||method=index||indexName={}||id={}||source={}||errMsg=index doc error. ",
+                    "method=index||indexName={}||id={}||source={}||response={}||errMsg=index failed",
-                    indexName, id, source, e);
+                    indexName, id, source, ConvertUtil.obj2Json(response), e
-            if (response != null) {
+            );
-                LOGGER.warn(
-                        "class=ESOpClient||method=index||indexName={}||id={}||source={}||errMsg=response {}",
-                        indexName, id, source, JSON.toJSONString(response));
-            }
-        } finally {
-            if (esClient != null) {
-                returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -271,19 +236,15 @@ public class ESOpClient {
 
     /**
      * 批量写入
-     *
-     * @param indexName
-     * @return
      */
     public boolean batchInsert(String indexName, List<? extends BaseESPO> pos) {
         if (CollectionUtils.isEmpty(pos)) {
             return true;
         }
 
-        ESClient esClient = null;
         ESBatchResponse response = null;
         try {
-            esClient = getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
             if (esClient == null) {
                 return false;
             }
@@ -312,16 +273,10 @@ public class ESOpClient {
                 return response.getRestStatus().getStatus() == HttpStatus.SC_OK && !response.getErrors();
             }
         } catch (Exception e) {
-            LOGGER.warn(
+            LOGGER.error(
-                    "method=batchInsert||indexName={}||errMsg=batch insert error. ", indexName, e);
+                    "method=batchInsert||indexName={}||response={}||errMsg=batch index failed",
-            if (response != null) {
+                    indexName, ConvertUtil.obj2Json(response), e
-                LOGGER.warn("method=batchInsert||indexName={}||errMsg=response {}", indexName, JSON.toJSONString(response));
+            );
-            }
-
-        } finally {
-            if (esClient != null) {
-                returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -331,9 +286,8 @@ public class ESOpClient {
      * 根据表达式判断索引是否已存在
      */
     public boolean indexExist(String indexName) {
-        ESClient esClient = null;
         try {
-            esClient = this.getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
             if (esClient == null) {
                 return false;
             }
@@ -341,11 +295,7 @@ public class ESOpClient {
             // 检查索引是否存在
             return esClient.admin().indices().prepareExists(indexName).execute().actionGet(30, TimeUnit.SECONDS).isExists();
         } catch (Exception e){
-            LOGGER.warn("class=ESOpClient||method=indexExist||indexName={}||msg=exception!", indexName, e);
+            LOGGER.error("method=indexExist||indexName={}||msg=exception!", indexName, e);
-        } finally {
-            if (esClient != null) {
-                returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -355,48 +305,45 @@ public class ESOpClient {
      * 创建索引
      */
     public boolean createIndex(String indexName) {
-        if (indexExist(indexName)) {
+        if (this.indexExist(indexName)) {
             return true;
         }
 
-        ESClient client = getESClientFromPool();
+        ESClient client = this.getESClientFromPool();
-        if (client != null) {
+        try {
-            try {
+            ESIndicesPutIndexResponse response = client
-                ESIndicesPutIndexResponse response = client.admin().indices().preparePutIndex(indexName).execute()
+                    .admin()
-                        .actionGet(ES_OPERATE_TIMEOUT, TimeUnit.SECONDS);
+                    .indices()
-                return response.getAcknowledged();
+                    .preparePutIndex(indexName)
-            } catch (Exception e){
+                    .execute()
-                LOGGER.warn( "msg=create index fail||indexName={}", indexName, e);
+                    .actionGet(ES_OPERATE_TIMEOUT, TimeUnit.SECONDS);
-            } finally {
+
-                returnESClientToPool(client);
+            return response.getAcknowledged();
-            }
+        } catch (Exception e){
+            LOGGER.error( "method=createIndex||indexName={}||errMsg=exception!", indexName, e);
         }
 
         return false;
     }
 
     public boolean templateExist(String indexTemplateName){
-        ESClient esClient = null;
-
         try {
-            esClient = this.getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
 
             // 获取es中原来index template的配置
-            ESIndicesGetTemplateResponse getTemplateResponse =
+            ESIndicesGetTemplateResponse getTemplateResponse = esClient
-                    esClient.admin().indices().prepareGetTemplate( indexTemplateName ).execute().actionGet( ES_OPERATE_TIMEOUT, TimeUnit.SECONDS );
+                    .admin()
+                    .indices()
+                    .prepareGetTemplate(indexTemplateName)
+                    .execute()
+                    .actionGet( ES_OPERATE_TIMEOUT, TimeUnit.SECONDS );
 
             TemplateConfig templateConfig = getTemplateResponse.getMultiTemplatesConfig().getSingleConfig();
-
             if (null != templateConfig) {
                 return true;
             }
         } catch (Exception e) {
-            LOGGER.warn( "method=templateExist||indexTemplateName={}||msg=exception!",
+            LOGGER.error( "method=templateExist||indexTemplateName={}||msg=exception!", indexTemplateName, e);
-                    indexTemplateName, e);
-        } finally {
-            if (esClient != null) {
-                this.returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -406,27 +353,29 @@ public class ESOpClient {
      * 创建索引模板
      */
     public boolean createIndexTemplateIfNotExist(String indexTemplateName, String config) {
-        ESClient esClient = null;
-
         try {
-            esClient = this.getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
 
-            //存在模板就返回，不存在就创建
+            // 存在模板就返回，不存在就创建
-            if(templateExist(indexTemplateName)){return true;}
+            if(this.templateExist(indexTemplateName)) {
+                return true;
+            }
 
             // 创建新的模板
-            ESIndicesPutTemplateResponse response = esClient.admin().indices().preparePutTemplate( indexTemplateName )
+            ESIndicesPutTemplateResponse response = esClient
-                    .setTemplateConfig( config ).execute().actionGet( ES_OPERATE_TIMEOUT, TimeUnit.SECONDS );
+                    .admin()
+                    .indices()
+                    .preparePutTemplate( indexTemplateName )
+                    .setTemplateConfig(config)
+                    .execute()
+                    .actionGet(ES_OPERATE_TIMEOUT, TimeUnit.SECONDS);
 
             return response.getAcknowledged();
         } catch (Exception e) {
-            LOGGER.warn( "method=createIndexTemplateIfNotExist||indexTemplateName={}||config={}||msg=exception!",
+            LOGGER.error(
+                    "method=createIndexTemplateIfNotExist||indexTemplateName={}||config={}||msg=exception!",
                     indexTemplateName, config, e
             );
-        } finally {
-            if (esClient != null) {
-                this.returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -434,54 +383,47 @@ public class ESOpClient {
 
     /**
      * 根据索引模板获取所有的索引
-     * @param indexName
-     * @return
      */
-    public List<String> listIndexByName(String indexName){
+    public List<String> listIndexByName(String indexName) {
-        ESClient esClient = null;
-
         try {
-            esClient = this.getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
 
-            ESIndicesCatIndicesResponse response = esClient.admin().indices().prepareCatIndices(indexName + "*").execute()
+            ESIndicesCatIndicesResponse response = esClient
+                    .admin()
+                    .indices()
+                    .prepareCatIndices(indexName + "*")
+                    .execute()
                     .actionGet(ES_OPERATE_TIMEOUT, TimeUnit.SECONDS);
-
+            if(null != response) {
-            if(null != response){
                 return response.getCatIndexResults().stream().map(CatIndexResult::getIndex).collect(Collectors.toList());
             }
         } catch (Exception e) {
-            LOGGER.warn( "method=listIndexByTemplate||indexName={}||msg=exception!",
+            LOGGER.error( "method=listIndexByName||indexName={}||msg=exception!", indexName, e);
-                    indexName, e);
-        } finally {
-            if (esClient != null) {
-                this.returnESClientToPool(esClient);
-            }
         }
 
-        return new ArrayList<>();
+        return Collections.emptyList();
     }
 
     /**
      * 删除索引
-     * @param indexRealName
-     * @return
      */
     public boolean delIndexByName(String indexRealName){
-        ESClient esClient = null;
-
         try {
-            esClient = this.getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
 
-            ESIndicesDeleteIndexResponse response = esClient.admin().indices().prepareDeleteIndex(indexRealName).execute()
+            ESIndicesDeleteIndexResponse response = esClient
+                    .admin()
+                    .indices()
+                    .prepareDeleteIndex(indexRealName)
+                    .execute()
                     .actionGet(ES_OPERATE_TIMEOUT, TimeUnit.SECONDS);
+
             return response.getAcknowledged();
+        } catch (ESIndexNotFoundException nfe) {
+            // 索引不存在时，debug环境时再进行打印
+            LOGGER.debug( "method=delIndexByName||indexRealName={}||errMsg=exception!", indexRealName, nfe);
         } catch (Exception e) {
-            LOGGER.warn( "method=delIndexByName||indexRealName={}||msg=exception!",
+            LOGGER.error( "method=delIndexByName||indexRealName={}||errMsg=exception!", indexRealName, e);
-                    indexRealName, e);
-        } finally {
-            if (esClient != null) {
-                this.returnESClientToPool(esClient);
-            }
         }
 
         return false;
@@ -491,61 +433,55 @@ public class ESOpClient {
 
     /**
      * 执行查询
-     * @param request
-     * @return
      */
     @Nullable
     private ESQueryResponse doQuery(ESQueryRequest request) {
-        ESClient esClient = null;
         try {
-            esClient = getESClientFromPool();
+            ESClient esClient = this.getESClientFromPool();
             ESQueryResponse response = esClient.query(request).actionGet(120, TimeUnit.SECONDS);
 
-            if(!EnvUtil.isOnline()){
+            LOGGER.debug(
-                LOGGER.info("method=doQuery||indexName={}||queryDsl={}||ret={}",
+                    "method=doQuery||indexName={}||queryDsl={}||ret={}",
-                        request.indices(), bytesReferenceConvertDsl(request.source()), JSON.toJSONString(response));
+                    request.indices(), bytesReferenceConvertDsl(request.source()), JSON.toJSONString(response)
-            }
+            );
 
             return response;
         } catch (Exception e) {
             LOGGER.error( "method=doQuery||indexName={}||queryDsl={}||errMsg=query error. ",
                     request.indices(), bytesReferenceConvertDsl(request.source()), e);
             return null;
-        }finally {
-            if (esClient != null) {
-                returnESClientToPool(esClient);
-            }
         }
     }
 
     private boolean handleErrorResponse(String indexName, List<? extends BaseESPO> pos, ESBatchResponse response) {
-        if (response.getErrors().booleanValue()) {
+        if (response.getErrors()) {
-            int errorItemIndex = 0;
+            return false;
-
-            if (CollectionUtils.isNotEmpty(response.getItems())) {
-                for (IndexResultItemNode item : response.getItems()) {
-                    recordErrorResponseItem(indexName, pos, errorItemIndex++, item);
-                }
-            }
-
-            return true;
         }
 
-        return false;
+        int errorItemIndex = 0;
+
+        if (CollectionUtils.isNotEmpty(response.getItems())) {
+            for (IndexResultItemNode item : response.getItems()) {
+                recordErrorResponseItem(indexName, pos, errorItemIndex++, item);
+            }
+        }
+
+        return true;
     }
 
     private void recordErrorResponseItem(String indexName, List<? extends BaseESPO> pos, int errorItemIndex, IndexResultItemNode item) {
-        if (item.getIndex() != null && item.getIndex().getShards() != null
+        if (item.getIndex() != null
+                && item.getIndex().getShards() != null
                 && CollectionUtils.isNotEmpty(item.getIndex().getShards().getFailures())) {
             LOGGER.warn(
-                    "class=ESOpClient||method=batchInsert||indexName={}||errMsg=Failures: {}, content: {}",
+                    "method=batchInsert||indexName={}||errMsg=Failures: {}, content: {}",
                     indexName, item.getIndex().getShards().getFailures().toString(),
                     JSON.toJSONString(pos.get(errorItemIndex)));
         }
 
         if (item.getIndex() != null && item.getIndex().getError() != null) {
             LOGGER.warn(
-                    "class=ESOpClient||method=batchInsert||indexName={}||errMsg=Error: {}, content: {}",
+                    "method=batchInsert||indexName={}||errMsg=Error: {}, content: {}",
                     indexName, item.getIndex().getError().getReason(),
                     JSON.toJSONString(pos.get(errorItemIndex)));
         }
@@ -553,21 +489,18 @@ public class ESOpClient {
 
     /**
      * 转换dsl语句
-     *
-     * @param bytes
-     * @return
      */
     private String bytesReferenceConvertDsl(BytesReference bytes) {
         try {
             return XContentHelper.convertToJson(bytes, false);
         } catch (IOException e) {
-            LOGGER.warn("class=ESOpClient||method=bytesReferenceConvertDsl||errMsg=fail to covert", e);
+            LOGGER.warn("method=bytesReferenceConvertDsl||errMsg=fail to covert", e);
         }
 
         return "";
     }
 
-    private ESClient buildEsClient(String address,String password,String clusterName, String version) {
+    private ESClient buildEsClient(String address, String password,String clusterName, String version) {
         if (StringUtils.isBlank(address)) {
             return null;
         }
@@ -602,7 +535,7 @@ public class ESOpClient {
                 // ignore
             }
 
-            LOGGER.error("class=ESESOpClient||method=buildEsClient||errMsg={}||address={}", e.getMessage(), address, e);
+            LOGGER.error("method=buildEsClient||address={}||errMsg=exception", address, e);
             return null;
         }
     }

km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/dao/TopicMetricESDAO.java

@@ -207,17 +207,27 @@ public class TopicMetricESDAO extends BaseMetricESDAO {
     /**
      * 获取每个 metric 的 topN 个 topic 的指标，如果获取不到 topN 的topics, 则默认返回 defaultTopics 的指标
      */
-    public Table<String/*metric*/, String/*topics*/, List<MetricPointVO>> listTopicMetricsByTopN(Long clusterPhyId, List<String> defaultTopics,
+    public Table<String/*metric*/, String/*topics*/, List<MetricPointVO>> listTopicMetricsByTopN(Long clusterPhyId,
-                                                                                                 List<String> metrics, String aggType, int topN,
+                                                                                                 List<String> defaultTopics,
-                                                                                                 Long startTime, Long endTime){
+                                                                                                 List<String> metrics,
+                                                                                                 String aggType,
+                                                                                                 int topN,
+                                                                                                 Long startTime,
+                                                                                                 Long endTime){
         //1、获取topN要查询的topic，每一个指标的topN的topic可能不一样
-        Map<String, List<String>> metricTopics = getTopNTopics(clusterPhyId, metrics, aggType, topN, startTime, endTime);
+        Map<String, List<String>> metricTopics = this.getTopNTopics(clusterPhyId, metrics, aggType, topN, startTime, endTime);
 
         Table<String, String, List<MetricPointVO>> table = HashBasedTable.create();
 
-        for(String metric : metrics){
+        for(String metric : metrics) {
-            table.putAll(listTopicMetricsByTopics(clusterPhyId, Arrays.asList(metric),
+            table.putAll(this.listTopicMetricsByTopics(
-                    aggType, metricTopics.getOrDefault(metric, defaultTopics), startTime, endTime));
+                    clusterPhyId,
+                    Arrays.asList(metric),
+                    aggType,
+                    metricTopics.getOrDefault(metric, defaultTopics),
+                    startTime,
+                    endTime)
+            );
         }
 
         return table;
@@ -226,9 +236,12 @@ public class TopicMetricESDAO extends BaseMetricESDAO {
     /**
      * 获取每个 metric 指定个 topic 的指标
      */
-    public Table<String/*metric*/, String/*topics*/, List<MetricPointVO>> listTopicMetricsByTopics(Long clusterPhyId, List<String> metrics,
+    public Table<String/*metric*/, String/*topics*/, List<MetricPointVO>> listTopicMetricsByTopics(Long clusterPhyId,
-                                                                                                      String aggType, List<String> topics,
+                                                                                                   List<String> metrics,
-                                                                                                      Long startTime, Long endTime){
+                                                                                                   String aggType,
+                                                                                                   List<String> topics,
+                                                                                                   Long startTime,
+                                                                                                   Long endTime){
         //1、获取需要查下的索引
         String realIndex = realIndex(startTime, endTime);
 

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/service/listener/TaskClusterAddedListener.java

@@ -22,7 +22,7 @@ public class TaskClusterAddedListener implements ApplicationListener<ClusterPhyA
 
     @Override
     public void onApplicationEvent(ClusterPhyAddedEvent event) {
-        LOGGER.info("class=TaskClusterAddedListener||method=onApplicationEvent||clusterPhyId={}||msg=listened new cluster", event.getClusterPhyId());
+        LOGGER.info("method=onApplicationEvent||clusterPhyId={}||msg=listened new cluster", event.getClusterPhyId());
         Long now = System.currentTimeMillis();
 
         // 交由KS自定义的线程池，异步执行任务