修复上报夜莺功能

kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/common/entry/N9eResult.java

+package com.xiaojukeji.kafka.manager.account.common.entry;
+
+/**
+ * @author zengqiao
+ * @date 20/9/7
+ */
+public class N9eResult<T> {
+    private T dat;
+
+    private String err;
+
+    public T getDat() {
+        return dat;
+    }
+
+    public void setDat(T dat) {
+        this.dat = dat;
+    }
+
+    public String getErr() {
+        return err;
+    }
+
+    public void setErr(String err) {
+        this.err = err;
+    }
+
+    @Override
+    public String toString() {
+        return "N9eResult{" +
+                "dat=" + dat +
+                ", err='" + err + '\'' +
+                '}';
+    }
+}
\ No newline at end of file

kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/common/entry/N9eUserData.java

+package com.xiaojukeji.kafka.manager.account.common.entry;
+
+/**
+ * @author zengqiao
+ * @date 20/9/7
+ */
+public class N9eUserData {
+    private Long id;
+
+    private String uuid;
+
+    private String username;
+
+    private String dispname;
+
+    private String phone;
+
+    private String email;
+
+    private String im;
+
+    private String portrait;
+
+    private Integer is_root;
+
+    private Integer leader_id;
+
+    private String leader_name;
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public String getUuid() {
+        return uuid;
+    }
+
+    public void setUuid(String uuid) {
+        this.uuid = uuid;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getDispname() {
+        return dispname;
+    }
+
+    public void setDispname(String dispname) {
+        this.dispname = dispname;
+    }
+
+    public String getPhone() {
+        return phone;
+    }
+
+    public void setPhone(String phone) {
+        this.phone = phone;
+    }
+
+    public String getEmail() {
+        return email;
+    }
+
+    public void setEmail(String email) {
+        this.email = email;
+    }
+
+    public String getIm() {
+        return im;
+    }
+
+    public void setIm(String im) {
+        this.im = im;
+    }
+
+    public String getPortrait() {
+        return portrait;
+    }
+
+    public void setPortrait(String portrait) {
+        this.portrait = portrait;
+    }
+
+    public Integer getIs_root() {
+        return is_root;
+    }
+
+    public void setIs_root(Integer is_root) {
+        this.is_root = is_root;
+    }
+
+    public Integer getLeader_id() {
+        return leader_id;
+    }
+
+    public void setLeader_id(Integer leader_id) {
+        this.leader_id = leader_id;
+    }
+
+    public String getLeader_name() {
+        return leader_name;
+    }
+
+    public void setLeader_name(String leader_name) {
+        this.leader_name = leader_name;
+    }
+
+    @Override
+    public String toString() {
+        return "EPRIResult{" +
+                "id=" + id +
+                ", uuid='" + uuid + '\'' +
+                ", username='" + username + '\'' +
+                ", dispname='" + dispname + '\'' +
+                ", phone='" + phone + '\'' +
+                ", email='" + email + '\'' +
+                ", im='" + im + '\'' +
+                ", portrait='" + portrait + '\'' +
+                ", is_root=" + is_root +
+                ", leader_id=" + leader_id +
+                ", leader_name='" + leader_name + '\'' +
+                '}';
+    }
+}
\ No newline at end of file

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/MonitorSinkConstant.java

@@ -5,8 +5,6 @@ package com.xiaojukeji.kafka.manager.monitor.common;
  * @date 17/5/24.
  */
 public class MonitorSinkConstant {
-    public static final String MONITOR_SYSTEM_TAG_DEFAULT_HOST = "kafka-manager-ser01.ys01";
-
     /**
      * 指标名前缀
      */

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/MetricSinkPoint.java

 package com.xiaojukeji.kafka.manager.monitor.common.entry;
 
-import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorBaseSinkTag;
-
-import java.text.DecimalFormat;
+import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.AbstractMonitorSinkTag;
 
 /**
  * @author huangjw
@@ -17,7 +15,7 @@ public class MetricSinkPoint {
     /**
      * 指标值
      */
-    private String value;
+    private Double value;
 
     /**
      * 上报周期
@@ -32,16 +30,11 @@ public class MetricSinkPoint {
     /**
      * tags
      */
-    private MonitorBaseSinkTag tags;
-
-    private static final DecimalFormat DECIMAL_FORMAT = new DecimalFormat("0.00");
+    private AbstractMonitorSinkTag tags;
 
-    public MetricSinkPoint(String name, double value, int step, long timestamp, MonitorBaseSinkTag tags) {
-        if (value <= 0.001) {
-            value = 0.0;
-        }
+    public MetricSinkPoint(String name, double value, int step, long timestamp, AbstractMonitorSinkTag tags) {
         this.name = name;
-        this.value = DECIMAL_FORMAT.format(value);
+        this.value = value;
         this.step = step;
         this.timestamp = timestamp;
         this.tags = tags;
@@ -51,25 +44,45 @@ public class MetricSinkPoint {
         return name;
     }
 
-    public String getValue() {
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Double getValue() {
         return value;
     }
 
+    public void setValue(Double value) {
+        this.value = value;
+    }
+
+    public int getStep() {
+        return step;
+    }
+
+    public void setStep(int step) {
+        this.step = step;
+    }
+
     public long getTimestamp() {
         return timestamp;
     }
 
-    public MonitorBaseSinkTag getTags() {
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public AbstractMonitorSinkTag getTags() {
         return tags;
     }
 
-    public int getStep() {
-        return step;
+    public void setTags(AbstractMonitorSinkTag tags) {
+        this.tags = tags;
     }
 
     @Override
     public String toString() {
-        return "MetricPoint{" +
+        return "MetricSinkPoint{" +
                 "name='" + name + '\'' +
                 ", value='" + value + '\'' +
                 ", timestamp=" + timestamp +

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/AbstractMonitorKafkaSinkTag.java

+package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
+
+/**
+ * @author zengqiao
+ * @date 20/5/24
+ */
+public abstract class AbstractMonitorKafkaSinkTag extends AbstractMonitorSinkTag {
+    protected String cluster;
+
+    public AbstractMonitorKafkaSinkTag(String cluster) {
+        this.cluster = cluster;
+    }
+}
\ No newline at end of file

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/AbstractMonitorSinkTag.java

+package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
+
+/**
+ * @author zengqiao
+ * @date 20/5/24
+ */
+public abstract class AbstractMonitorSinkTag {
+    public abstract String convert2Tags();
+}
\ No newline at end of file

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorBaseSinkTag.java

-package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
-
-/**
- * @author zengqiao
- * @date 20/5/24
- */
-public class MonitorBaseSinkTag {
-    /**
-     * 主机名
-     */
-    protected String host;
-
-    public MonitorBaseSinkTag(String host) {
-        this.host = host;
-    }
-
-    public String getHost() {
-        return host;
-    }
-
-    public void setHost(String host) {
-        this.host = host;
-    }
-
-    @Override
-    public String toString() {
-        return "BaseTag{" +
-                "host='" + host + '\'' +
-                '}';
-    }
-}
\ No newline at end of file

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorConsumePartitionSinkTag.java

@@ -4,19 +4,11 @@ package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
  * @author zengqiao
  * @date 20/09/02
  */
-public class MonitorConsumePartitionSinkTag extends MonitorTopicPartitionSinkTag {
+public class MonitorConsumePartitionSinkTag extends MonitorPartitionSinkTag {
     private String consumerGroup;
 
-    public MonitorConsumePartitionSinkTag(String host, String cluster, String topic, Integer partition, String consumerGroup) {
-        super(host, cluster, topic, partition);
-        this.consumerGroup = consumerGroup;
-    }
-
-    public String getConsumerGroup() {
-        return consumerGroup;
-    }
-
-    public void setConsumerGroup(String consumerGroup) {
+    public MonitorConsumePartitionSinkTag(String cluster, String topic, Integer partition, String consumerGroup) {
+        super(cluster, topic, partition);
         this.consumerGroup = consumerGroup;
     }
 
@@ -27,7 +19,11 @@ public class MonitorConsumePartitionSinkTag extends MonitorTopicPartitionSinkTag
                 ", topic='" + topic + '\'' +
                 ", partition='" + partition + '\'' +
                 ", cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
                 '}';
     }
+
+    @Override
+    public String convert2Tags() {
+        return String.format("cluster=%s,topic=%s,partition=%s,consumerGroup=%s", cluster, topic, partition, consumerGroup);
+    }
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorConsumerSinkTag.java → kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorConsumeTopicSinkTag.java

@@ -4,40 +4,25 @@ package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
  * @author huangjw
  * @date 17/5/25.
  */
-public class MonitorConsumerSinkTag extends MonitorKafkaBaseSinkTag {
-    private String topic;
-
+public class MonitorConsumeTopicSinkTag extends MonitorTopicSinkTag {
     private String consumerGroup;
 
-    public MonitorConsumerSinkTag(String host, String cluster, String topic, String consumerGroup) {
-        super(host, cluster);
-        this.topic = topic;
-        this.consumerGroup = consumerGroup;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
-        this.topic = topic;
-    }
-
-    public String getConsumerGroup() {
-        return consumerGroup;
-    }
-
-    public void setConsumerGroup(String consumerGroup) {
+    public MonitorConsumeTopicSinkTag(String cluster, String topic, String consumerGroup) {
+        super(cluster, topic);
         this.consumerGroup = consumerGroup;
     }
 
     @Override
     public String toString() {
-        return "MonitorConsumerSinkTag{" +
-                "topic='" + topic + '\'' +
-                ", consumerGroup='" + consumerGroup + '\'' +
+        return "MonitorConsumeTopicSinkTag{" +
+                "consumerGroup='" + consumerGroup + '\'' +
+                ", topic='" + topic + '\'' +
                 ", cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
                 '}';
     }
+
+    @Override
+    public String convert2Tags() {
+        return String.format("cluster=%s,topic=%s,consumerGroup=%s", cluster, topic, consumerGroup);
+    }
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorKafkaBaseSinkTag.java

-package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
-
-/**
- * @author zengqiao
- * @date 20/5/24
- */
-public class MonitorKafkaBaseSinkTag extends MonitorBaseSinkTag {
-    protected String cluster;
-
-    public MonitorKafkaBaseSinkTag(String host, String cluster) {
-        super(host);
-        this.cluster = cluster;
-    }
-
-    public String getCluster() {
-        return cluster;
-    }
-
-    public void setCluster(String cluster) {
-        this.cluster = cluster;
-    }
-
-    @Override
-    public String toString() {
-        return "MonitorKafkaBaseSinkTag{" +
-                "cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
-                '}';
-    }
-}
\ No newline at end of file

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorTopicPartitionSinkTag.java → kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorPartitionSinkTag.java

@@ -4,40 +4,26 @@ package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
  * @author zengqiao
  * @date 20/9/3
  */
-public class MonitorTopicPartitionSinkTag extends MonitorKafkaBaseSinkTag {
-    protected String topic;
-
+public class MonitorPartitionSinkTag extends MonitorTopicSinkTag {
     protected String partition;
 
-    public MonitorTopicPartitionSinkTag(String host, String cluster, String topic, Integer partition) {
-        super(host, cluster);
+    public MonitorPartitionSinkTag(String cluster, String topic, Integer partition) {
+        super(cluster, topic);
         this.topic = topic;
         this.partition = String.valueOf(partition);
     }
 
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
-        this.topic = topic;
-    }
-
-    public String getPartition() {
-        return partition;
-    }
-
-    public void setPartition(String partition) {
-        this.partition = partition;
-    }
-
     @Override
     public String toString() {
-        return "MonitorTopicPartitionSinkTag{" +
+        return "MonitorPartitionSinkTag{" +
                 "topic='" + topic + '\'' +
                 ", partition='" + partition + '\'' +
                 ", cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
                 '}';
     }
+
+    @Override
+    public String convert2Tags() {
+        return String.format("cluster=%s,topic=%s,partition=%s", cluster, topic, partition);
+    }
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorTopicSinkTag.java

@@ -4,20 +4,11 @@ package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
  * @author huangjw
  * @date 17/5/24.
  */
-public class MonitorTopicSinkTag extends MonitorKafkaBaseSinkTag {
+public class MonitorTopicSinkTag extends AbstractMonitorKafkaSinkTag {
+    protected String topic;
 
-    private String topic;
-
-    public MonitorTopicSinkTag(String host, String cluster, String topic) {
-        super(host, cluster);
-        this.topic = topic;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
+    public MonitorTopicSinkTag(String cluster, String topic) {
+        super(cluster);
         this.topic = topic;
     }
 
@@ -26,7 +17,11 @@ public class MonitorTopicSinkTag extends MonitorKafkaBaseSinkTag {
         return "MonitorTopicSinkTag{" +
                 "topic='" + topic + '\'' +
                 ", cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
                 '}';
     }
+
+    @Override
+    public String convert2Tags() {
+        return String.format("cluster=%s,topic=%s", cluster, topic);
+    }
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/common/entry/sink/MonitorTopicThrottledSinkTag.java

@@ -4,40 +4,16 @@ package com.xiaojukeji.kafka.manager.monitor.common.entry.sink;
  * @author crius
  * @date 19/5/21.
  */
-public class MonitorTopicThrottledSinkTag extends MonitorKafkaBaseSinkTag {
-    private String topic;
-
+public class MonitorTopicThrottledSinkTag extends MonitorTopicSinkTag {
     private String appId;
 
-    public MonitorTopicThrottledSinkTag(String host, String cluster, String topic, String appId) {
-        super(host, cluster);
-        this.topic = topic;
-        this.appId = appId;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
-        this.topic = topic;
-    }
-
-    public String getAppId() {
-        return appId;
-    }
-
-    public void setAppId(String appId) {
+    public MonitorTopicThrottledSinkTag(String cluster, String topic, String appId) {
+        super(cluster, topic);
         this.appId = appId;
     }
 
     @Override
-    public String toString() {
-        return "MonitorTopicThrottledSinkTag{" +
-                "topic='" + topic + '\'' +
-                ", appId='" + appId + '\'' +
-                ", cluster='" + cluster + '\'' +
-                ", host='" + host + '\'' +
-                '}';
+    public String convert2Tags() {
+        return String.format("cluster=%s,topic=%s,appId=%s", cluster, topic, appId);
     }
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/component/n9e/N9eConverter.java

@@ -13,13 +13,14 @@ import java.util.List;
  * @date 20/8/26
  */
 public class N9eConverter {
-    public static List<N9eMetricSinkPoint> convert2N9eMetricSinkPointList(List<MetricSinkPoint> pointList) {
+    public static List<N9eMetricSinkPoint> convert2N9eMetricSinkPointList(String nid, List<MetricSinkPoint> pointList) {
         if (pointList == null || pointList.isEmpty()) {
             return new ArrayList<>();
         }
         List<N9eMetricSinkPoint> n9ePointList = new ArrayList<>();
         for (MetricSinkPoint sinkPoint: pointList) {
             n9ePointList.add(new N9eMetricSinkPoint(
+                    nid,
                     sinkPoint.getName(),
                     sinkPoint.getValue(),
                     sinkPoint.getStep(),

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/component/n9e/N9eService.java

@@ -28,10 +28,13 @@ public class N9eService extends AbstractMonitorService {
     private Integer monitorN9eNid;
 
     @Value("${monitor.n9e.user-token}")
-    private String monitorN9eToken;
+    private String monitorN9eUserToken;
 
-    @Value("${monitor.n9e.base-url}")
-    private String monitorN9eBaseUrl;
+    @Value("${monitor.n9e.mon.base-url}")
+    private String monitorN9eMonBaseUrl;
+
+    @Value("${monitor.n9e.sink.base-url}")
+    private String monitorN9eSinkBaseUrl;
 
     /**
      * 告警策略
@@ -82,7 +85,7 @@ public class N9eService extends AbstractMonitorService {
         String response = null;
         try {
             response = HttpUtils.postForString(
-                    monitorN9eBaseUrl + STRATEGY_ADD_URL,
+                    monitorN9eMonBaseUrl + STRATEGY_ADD_URL,
                     JSON.toJSONString(N9eConverter.convert2N9eStrategy(strategy, monitorN9eNid)),
                     buildHeader()
             );
@@ -106,7 +109,7 @@ public class N9eService extends AbstractMonitorService {
         String response = null;
         try {
             response = HttpUtils.deleteForString(
-                    monitorN9eBaseUrl + STRATEGY_DEL_URL,
+                    monitorN9eMonBaseUrl + STRATEGY_DEL_URL,
                     JSON.toJSONString(params),
                     buildHeader()
             );
@@ -127,7 +130,7 @@ public class N9eService extends AbstractMonitorService {
         String response = null;
         try {
             response = HttpUtils.putForString(
-                    monitorN9eBaseUrl + STRATEGY_MODIFY_URL,
+                    monitorN9eMonBaseUrl + STRATEGY_MODIFY_URL,
                     JSON.toJSONString(N9eConverter.convert2N9eStrategy(strategy, monitorN9eNid)),
                     buildHeader()
             );
@@ -150,7 +153,7 @@ public class N9eService extends AbstractMonitorService {
 
         String response = null;
         try {
-            response = HttpUtils.get(monitorN9eBaseUrl + STRATEGY_QUERY_BY_NS_URL, params, buildHeader());
+            response = HttpUtils.get(monitorN9eMonBaseUrl + STRATEGY_QUERY_BY_NS_URL, params, buildHeader());
             N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
             if (!ValidateUtils.isBlank(n9eResult.getErr())) {
                 LOGGER.error("get monitor strategies failed, response:{}.", response);
@@ -169,7 +172,7 @@ public class N9eService extends AbstractMonitorService {
 
         String response = null;
         try {
-            response = HttpUtils.get(monitorN9eBaseUrl + uri, new HashMap<>(0), buildHeader());
+            response = HttpUtils.get(monitorN9eMonBaseUrl + uri, new HashMap<>(0), buildHeader());
             N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
             if (!ValidateUtils.isBlank(n9eResult.getErr())) {
                 LOGGER.error("get monitor strategy failed, response:{}.", response);
@@ -221,13 +224,13 @@ public class N9eService extends AbstractMonitorService {
     public Boolean sinkMetrics(List<MetricSinkPoint> pointList) {
         String response = null;
         try {
-            String content = JSON.toJSONString(N9eConverter.convert2N9eMetricSinkPointList(pointList));
+            String content = JSON.toJSONString(N9eConverter.convert2N9eMetricSinkPointList(String.valueOf(this.monitorN9eNid), pointList));
 
             long startTime = System.currentTimeMillis();
             response = HttpUtils.postForString(
-                    monitorN9eBaseUrl + COLLECTOR_SINK_DATA_URL,
+                    monitorN9eSinkBaseUrl + COLLECTOR_SINK_DATA_URL,
                     content,
-                    null
+                    buildHeader()
             );
             LOGGER.info("sinkMetrics cost-time:{}.", System.currentTimeMillis() - startTime);
 
@@ -251,7 +254,7 @@ public class N9eService extends AbstractMonitorService {
     public List<NotifyGroup> getNotifyGroups() {
         String response = null;
         try {
-            response = HttpUtils.get(monitorN9eBaseUrl + ALL_NOTIFY_GROUP_URL, new HashMap<>(0), buildHeader());
+            response = HttpUtils.get(monitorN9eMonBaseUrl + ALL_NOTIFY_GROUP_URL, new HashMap<>(0), buildHeader());
             N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
             if (!ValidateUtils.isBlank(n9eResult.getErr())) {
                 LOGGER.error("get notify group failed, response:{}.", response);
@@ -267,8 +270,7 @@ public class N9eService extends AbstractMonitorService {
     private Map<String, String> buildHeader() {
         Map<String, String> header = new HashMap<>(2);
         header.put("Content-Type", "application/json");
-        header.put("X-User-Token", monitorN9eToken);
+        header.put("X-User-Token", monitorN9eUserToken);
         return header;
     }
-
 }

kafka-manager-extends/kafka-manager-monitor/src/main/java/com/xiaojukeji/kafka/manager/monitor/component/n9e/entry/N9eMetricSinkPoint.java

 package com.xiaojukeji.kafka.manager.monitor.component.n9e.entry;
 
-import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorBaseSinkTag;
+import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.AbstractMonitorSinkTag;
 
 /**
  * 夜莺上报监控数据点
@@ -9,19 +9,19 @@ import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorBaseSinkTag
  */
 public class N9eMetricSinkPoint {
     /**
-     * 指标名
+     * 节点ID
      */
-    private String metric;
+    private String nid;
 
     /**
-     * 指标对应的机器
+     * 指标名
      */
-    private String endpoint;
+    private String metric;
 
     /**
      * 指标值
      */
-    private String value;
+    private Double value;
 
     /**
      * 上报周期
@@ -36,42 +36,43 @@ public class N9eMetricSinkPoint {
     /**
      * tags
      */
-    private MonitorBaseSinkTag tags;
+    private String tags;
 
-    public N9eMetricSinkPoint(String metric,
-                              String value,
+    public N9eMetricSinkPoint(String nid,
+                              String metric,
+                              Double value,
                               int step,
                               long timestamp,
-                              MonitorBaseSinkTag tags) {
+                              AbstractMonitorSinkTag tags) {
+        this.nid = nid;
         this.metric = metric;
-        this.endpoint = tags.getHost();
         this.value = value;
         this.step = step;
         this.timestamp = timestamp;
-        this.tags = tags;
+        this.tags = tags.convert2Tags();
     }
 
-    public String getMetric() {
-        return metric;
+    public String getNid() {
+        return nid;
     }
 
-    public void setMetric(String metric) {
-        this.metric = metric;
+    public void setNid(String nid) {
+        this.nid = nid;
     }
 
-    public String getEndpoint() {
-        return endpoint;
+    public String getMetric() {
+        return metric;
     }
 
-    public void setEndpoint(String endpoint) {
-        this.endpoint = endpoint;
+    public void setMetric(String metric) {
+        this.metric = metric;
     }
 
-    public String getValue() {
+    public Double getValue() {
         return value;
     }
 
-    public void setValue(String value) {
+    public void setValue(Double value) {
         this.value = value;
     }
 
@@ -91,23 +92,23 @@ public class N9eMetricSinkPoint {
         this.timestamp = timestamp;
     }
 
-    public MonitorBaseSinkTag getTags() {
+    public String getTags() {
         return tags;
     }
 
-    public void setTags(MonitorBaseSinkTag tags) {
+    public void setTags(String tags) {
         this.tags = tags;
     }
 
     @Override
     public String toString() {
         return "N9eMetricSinkPoint{" +
-                "metric='" + metric + '\'' +
-                ", endpoint='" + endpoint + '\'' +
-                ", value='" + value + '\'' +
+                "nid=" + nid +
+                ", metric='" + metric + '\'' +
+                ", value=" + value +
                 ", step=" + step +
                 ", timestamp=" + timestamp +
-                ", tags=" + tags +
+                ", tags='" + tags + '\'' +
                 '}';
     }
 }
\ No newline at end of file

kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/metrics/store/SinkCommunityTopicMetrics2Monitor.java

@@ -100,7 +100,6 @@ public class SinkCommunityTopicMetrics2Monitor extends AbstractScheduledTask<Clu
                         MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                         timestamp,
                         new MonitorTopicSinkTag(
-                                MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                                 logicalClusterName,
                                 metrics.getTopicName()
                         )
@@ -112,7 +111,6 @@ public class SinkCommunityTopicMetrics2Monitor extends AbstractScheduledTask<Clu
                         MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                         timestamp,
                         new MonitorTopicSinkTag(
-                                MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                                 logicalClusterName,
                                 metrics.getTopicName()
                         )
@@ -124,7 +122,6 @@ public class SinkCommunityTopicMetrics2Monitor extends AbstractScheduledTask<Clu
                         MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                         timestamp,
                         new MonitorTopicSinkTag(
-                                MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                                 logicalClusterName,
                                 metrics.getTopicName()
                         )

kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkConsumerMetrics2Monitor.java

@@ -11,7 +11,7 @@ import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
 import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
 import com.xiaojukeji.kafka.manager.monitor.common.entry.MetricSinkPoint;
 import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorConsumePartitionSinkTag;
-import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorConsumerSinkTag;
+import com.xiaojukeji.kafka.manager.monitor.common.entry.sink.MonitorConsumeTopicSinkTag;
 import com.xiaojukeji.kafka.manager.monitor.component.AbstractMonitorService;
 import com.xiaojukeji.kafka.manager.service.cache.KafkaMetricsCache;
 import com.xiaojukeji.kafka.manager.service.cache.LogicalClusterMetadataManager;
@@ -99,7 +99,6 @@ public class SinkConsumerMetrics2Monitor implements ApplicationListener<Consumer
                     MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                     timestamp,
                     new MonitorConsumePartitionSinkTag(
-                            MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                             logicalClusterName,
                             metrics.getTopicName(),
                             partitionId,
@@ -115,8 +114,7 @@ public class SinkConsumerMetrics2Monitor implements ApplicationListener<Consumer
                 maxLag,
                 MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                 timestamp,
-                new MonitorConsumerSinkTag(
-                        MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
+                new MonitorConsumeTopicSinkTag(
                         logicalClusterName,
                         metrics.getTopicName(),
                         metrics.getConsumerGroup()
@@ -139,8 +137,7 @@ public class SinkConsumerMetrics2Monitor implements ApplicationListener<Consumer
                 maxDelayTime,
                 MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                 timestamp,
-                new MonitorConsumerSinkTag(
-                        MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
+                new MonitorConsumeTopicSinkTag(
                         logicalClusterName,
                         metrics.getTopicName(),
                         metrics.getConsumerGroup()

kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkTopicThrottledMetrics2Monitor.java

@@ -85,7 +85,6 @@ public class SinkTopicThrottledMetrics2Monitor implements ApplicationListener<To
                     MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                     startTime / 1000,
                     new MonitorTopicThrottledSinkTag(
-                            MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                             logicalClusterName,
                             metrics.getTopicName(),
                             metrics.getAppId()
@@ -100,7 +99,6 @@ public class SinkTopicThrottledMetrics2Monitor implements ApplicationListener<To
                     MonitorSinkConstant.MONITOR_SYSTEM_SINK_STEP,
                     startTime / 1000,
                     new MonitorTopicThrottledSinkTag(
-                            MonitorSinkConstant.MONITOR_SYSTEM_TAG_DEFAULT_HOST,
                             logicalClusterName,
                             metrics.getTopicName(),
                             metrics.getAppId()

kafka-manager-web/src/main/resources/application.yml

@@ -19,7 +19,7 @@ spring:
     allow-bean-definition-overriding: true
 
   profiles:
-    active: dev,account-dev,monitor-dev,notify-dev,kcm-dev
+    active: dev
   servlet:
     multipart:
       max-file-size: 60MB
@@ -48,10 +48,12 @@ kcm:
 monitor:
   enabled: false
   n9e:
-    base-url: http://127.0.0.1/api
-    username: admin
-    user-token: admin
-    nid: 10
+    nid: 2
+    user-token: 1234567890
+    mon:
+      base-url: http://127.0.0.1:8032
+    sink:
+      base-url: http://127.0.0.1:8008
 
 notify:
   kafka: