[Feature]MM2管理-读写ES中的MM2指标(#894)

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/metrics/mm2/MirrorMakerMetrics.java

@@ -0,0 +1,46 @@
+package com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.mm2;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.BaseMetrics;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+/**
+ * @author zengqiao
+ * @date 20/6/17
+ */
+@Data
+@NoArgsConstructor
+@ToString
+public class MirrorMakerMetrics extends BaseMetrics {
+    private Long connectClusterId;
+
+    private String connectorName;
+
+    private String connectorNameAndClusterId;
+
+    public MirrorMakerMetrics(Long connectClusterId, String connectorName) {
+        super(null);
+        this.connectClusterId           = connectClusterId;
+        this.connectorName              = connectorName;
+        this.connectorNameAndClusterId  = connectorName + "#" + connectClusterId;
+    }
+
+    public MirrorMakerMetrics(Long clusterPhyId, Long connectClusterId, String connectorName) {
+        super(clusterPhyId);
+        this.connectClusterId           = connectClusterId;
+        this.connectorName              = connectorName;
+        this.connectorNameAndClusterId  = connectorName + "#" + connectClusterId;
+    }
+
+    public static MirrorMakerMetrics initWithMetric(Long connectClusterId, String connectorName, String metricName, Float value) {
+        MirrorMakerMetrics metrics = new MirrorMakerMetrics(connectClusterId, connectorName);
+        metrics.putMetric(metricName, value);
+        return metrics;
+    }
+
+    @Override
+    public String unique() {
+        return "KCOR@" + connectClusterId + "@" + connectorName;
+    }
+}

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/po/metrice/mm2/MirrorMakerMetricPO.java

@@ -0,0 +1,39 @@
+package com.xiaojukeji.know.streaming.km.common.bean.po.metrice.mm2;
+
+import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.BaseMetricESPO;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import static com.xiaojukeji.know.streaming.km.common.utils.CommonUtils.monitorTimestamp2min;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class MirrorMakerMetricPO extends BaseMetricESPO {
+    private Long        connectClusterId;
+
+    private String      connectorName;
+
+    /**
+     * 用于es内部排序
+     */
+    private String      connectorNameAndClusterId;
+
+    public MirrorMakerMetricPO(Long kafkaClusterPhyId, Long connectClusterId, String connectorName){
+        super(kafkaClusterPhyId);
+        this.connectClusterId               = connectClusterId;
+        this.connectorName                  = connectorName;
+        this.connectorNameAndClusterId      = connectorName + "#" + connectClusterId;
+    }
+
+    @Override
+    public String getKey() {
+        return "KCOR@" + clusterPhyId + "@" + connectClusterId + "@" + connectorName + "@" + monitorTimestamp2min(timestamp);
+    }
+
+    @Override
+    public String getRoutingValue() {
+        return String.valueOf(connectClusterId);
+    }
+}

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

@@ -13,27 +13,16 @@ import com.xiaojukeji.know.streaming.km.common.utils.Triple;
 import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
 import com.xiaojukeji.know.streaming.km.persistence.es.dao.BaseMetricESDAO;
 import com.xiaojukeji.know.streaming.km.persistence.es.dsls.DslConstant;
-import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
 import java.util.*;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 import static com.xiaojukeji.know.streaming.km.common.constant.ESConstant.*;
-import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CONNECT_CONNECTOR_INDEX;
 
-@Component
+public class BaseConnectorMetricESDAO extends BaseMetricESDAO {
-public class ConnectorMetricESDAO extends BaseMetricESDAO {
-
-    @PostConstruct
-    public void init() {
-        super.indexName     = CONNECT_CONNECTOR_INDEX;
-        checkCurrentDayIndexExist();
-        register( this);
-    }
 
     /**
-     * 获取每个 metric 的 topN 个 connector 的指标，如果获取不到 topN 的 connectors, 则默认返回 defaultTopics 的指标
+     * 获取每个 metric 的 topN 个 connector 的指标，如果获取不到 topN 的 connectors, 则默认返回 defaultConnectorList 的指标
      */
     public Table<String/*metric*/, Tuple<Long, String>, List<MetricPointVO>> listMetricsByTopN(Long clusterPhyId,
                                                                                                List<Tuple<Long, String>> defaultConnectorList,
@@ -143,7 +132,7 @@ public class ConnectorMetricESDAO extends BaseMetricESDAO {
         for(Tuple<Long, String> connector : connectorList) {
             try {
                 esTPService.submitSearchTask(
-                        String.format("class=ConnectorMetricESDAO||method=listMetricsByConnectors||ClusterPhyId=%d||connectorName=%s", clusterPhyId, connector.getV2()),
+                        String.format("class=BaseConnectorMetricESDAO||method=listMetricsByConnectors||ClusterPhyId=%d||connectorName=%s", clusterPhyId, connector.getV2()),
                         3000,
                         () -> {
                             String dsl = dslLoaderUtil.getFormatDslByFileName(
@@ -318,7 +307,9 @@ public class ConnectorMetricESDAO extends BaseMetricESDAO {
 
     private Tuple<String, Long> splitConnectorNameAndClusterId(String connectorNameAndClusterId){
         String[] ss = connectorNameAndClusterId.split("#");
-        if(null == ss || ss.length != 2){return null;}
+        if(null == ss || ss.length != 2) {
+            return null;
+        }
 
         return new Tuple<>(ss[0], Long.valueOf(ss[1]));
     }

km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/dao/connect/connector/ConnectorMetricESDAO.java

@@ -0,0 +1,19 @@
+package com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.connector;
+
+import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.BaseConnectorMetricESDAO;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CONNECT_CONNECTOR_INDEX;
+
+@Component
+public class ConnectorMetricESDAO extends BaseConnectorMetricESDAO {
+
+    @PostConstruct
+    public void init() {
+        super.indexName     = CONNECT_CONNECTOR_INDEX;
+        checkCurrentDayIndexExist();
+        register( this);
+    }
+}

km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/dao/connect/mm2/MirrorMakerMetricESDAO.java

@@ -0,0 +1,18 @@
+package com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.mm2;
+
+import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.BaseConnectorMetricESDAO;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CONNECT_MM2_INDEX;
+
+@Component
+public class MirrorMakerMetricESDAO extends BaseConnectorMetricESDAO {
+
+    @PostConstruct
+    public void init() {
+        super.indexName     = CONNECT_MM2_INDEX;
+        checkCurrentDayIndexExist();
+        register( this);
+    }
+}

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

@@ -89,6 +89,4 @@ public class DslConstant {
     public static final String GET_CONNECTOR_AGG_LIST_METRICS           = "ConnectorMetricESDAO/getConnectorAggListMetric";
 
     public static final String GET_CONNECTOR_AGG_TOP_METRICS            = "ConnectorMetricESDAO/getConnectorAggTopMetric";
-
-
 }

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

@@ -12,6 +12,7 @@ public class TemplateConstant {
     public static final String ZOOKEEPER_INDEX          = "ks_kafka_zookeeper_metric";
     public static final String CONNECT_CLUSTER_INDEX    = "ks_kafka_connect_cluster_metric";
     public static final String CONNECT_CONNECTOR_INDEX  = "ks_kafka_connect_connector_metric";
+    public static final String CONNECT_MM2_INDEX        = "ks_kafka_connect_mirror_maker_metric";
 
     private TemplateConstant() {
     }

km-persistence/src/main/resources/es/template/ks_kafka_connect_mirror_maker_metric

@@ -0,0 +1,98 @@
+{
+    "order" : 10,
+    "index_patterns" : [
+      "ks_kafka_connect_mirror_maker_metric*"
+    ],
+    "settings" : {
+      "index" : {
+        "number_of_shards" : "2"
+      }
+    },
+    "mappings" : {
+      "properties" : {
+        "connectClusterId" : {
+          "type" : "long"
+        },
+        "routingValue" : {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "ignore_above" : 256,
+              "type" : "keyword"
+            }
+          }
+        },
+        "connectorName" : {
+          "type" : "keyword"
+        },
+        "connectorNameAndClusterId" : {
+          "type" : "keyword"
+        },
+        "clusterPhyId" : {
+          "type" : "long"
+        },
+        "metrics" : {
+          "properties" : {
+            "HealthState" : {
+              "type" : "float"
+            },
+            "HealthCheckTotal" : {
+              "type" : "float"
+            },
+            "ByteCount" : {
+              "type" : "float"
+            },
+            "ByteRate" : {
+              "type" : "float"
+            },
+            "RecordAgeMs" : {
+              "type" : "float"
+            },
+            "RecordAgeMsAvg" : {
+              "type" : "float"
+            },
+            "RecordAgeMsMax" : {
+              "type" : "float"
+            },
+            "RecordAgeMsMin" : {
+              "type" : "float"
+            },
+            "RecordCount" : {
+              "type" : "float"
+            },
+            "RecordRate" : {
+              "type" : "float"
+            },
+            "ReplicationLatencyMs" : {
+              "type" : "float"
+            },
+            "ReplicationLatencyMsAvg" : {
+              "type" : "float"
+            },
+            "ReplicationLatencyMsMax" : {
+               "type" : "float"
+            },
+            "ReplicationLatencyMsMin" : {
+               "type" : "float"
+            }
+          }
+        },
+        "key" : {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "ignore_above" : 256,
+              "type" : "keyword"
+            }
+          }
+        },
+        "timestamp" : {
+          "format" : "yyyy-MM-dd HH:mm:ss Z||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS Z||yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss,SSS||yyyy/MM/dd HH:mm:ss||yyyy-MM-dd HH:mm:ss,SSS Z||yyyy/MM/dd HH:mm:ss,SSS Z||epoch_millis",
+          "index" : true,
+          "type" : "date",
+          "doc_values" : true
+        }
+      }
+    },
+    "aliases" : { }
+  }