[INLONG-2264] DataProxy get metric value with error JMX ObjectName (#2275)

* [INLONG-2264] DataProxy get metric value with error JMX ObjectName

* add UT for this bug

* Thread.sleep(2000);

Author: luchunliang

inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/MetricListenerRunnable.java

@@ -36,6 +36,8 @@
 import org.apache.inlong.commons.config.metrics.MetricItem;
 import org.apache.inlong.commons.config.metrics.MetricItemMBean;
 import org.apache.inlong.commons.config.metrics.MetricItemSetMBean;
+import org.apache.inlong.commons.config.metrics.MetricRegister;
+import org.apache.inlong.commons.config.metrics.MetricUtils;
 import org.apache.inlong.commons.config.metrics.MetricValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -46,7 +48,7 @@
  */
 public class MetricListenerRunnable implements Runnable {
 
-    public static final Logger LOG = LoggerFactory.getLogger(MetricObserver.class);
+    public static final Logger LOG = LoggerFactory.getLogger(MetricListenerRunnable.class);
 
     private String domain;
     private List<MetricListener> listenerList;
@@ -92,9 +94,14 @@ public void run() {
      * @throws ClassNotFoundException
      */
     @SuppressWarnings("unchecked")
-    private List<MetricItemValue> getItemValues() throws InstanceNotFoundException, AttributeNotFoundException,
+    public List<MetricItemValue> getItemValues() throws InstanceNotFoundException, AttributeNotFoundException,
             ReflectionException, MBeanException, MalformedObjectNameException, ClassNotFoundException {
-        ObjectName objName = new ObjectName(domain + MetricItemMBean.DOMAIN_SEPARATOR + "*");
+        StringBuilder beanName = new StringBuilder();
+        beanName.append(MetricRegister.JMX_DOMAIN).append(MetricItemMBean.DOMAIN_SEPARATOR)
+                .append("type=").append(MetricUtils.getDomain(DataProxyMetricItemSet.class))
+                .append(MetricItemMBean.PROPERTY_SEPARATOR)
+                .append("*");
+        ObjectName objName = new ObjectName(beanName.toString());
         final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
         Set<ObjectInstance> mbeans = mbs.queryMBeans(objName, null);
         LOG.info("getItemValues for domain:{},queryMBeans:{}", domain, mbeans);

inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/prometheus/PrometheusMetricListener.java

@@ -52,12 +52,12 @@
 import org.apache.inlong.dataproxy.metrics.DataProxyMetricItem;
 import org.apache.inlong.dataproxy.metrics.MetricItemValue;
 import org.apache.inlong.dataproxy.metrics.MetricListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.prometheus.client.Collector;
 import io.prometheus.client.CounterMetricFamily;
 import io.prometheus.client.exporter.HTTPServer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  *
@@ -70,10 +70,9 @@ public class PrometheusMetricListener extends Collector implements MetricListene
     public static final int DEFAULT_PROMETHEUS_HTTP_PORT = 8080;
     public static final String DEFAULT_DIMENSION_LABEL = "dimension";
 
-    //
+    private String metricName;
     private DataProxyMetricItem metricItem;
     private Map<String, AtomicLong> metricValueMap = new ConcurrentHashMap<>();
-    private String metricName;
     protected HTTPServer httpServer;
     private Map<String, MetricItemValue> dimensionMetricValueMap = new ConcurrentHashMap<>();
     private List<String> dimensionKeys = new ArrayList<>();
@@ -82,19 +81,20 @@ public class PrometheusMetricListener extends Collector implements MetricListene
      * Constructor
      */
     public PrometheusMetricListener() {
+        this.metricName = CommonPropertiesHolder.getString(RemoteConfigManager.KEY_PROXY_CLUSTER_NAME);
         this.metricItem = new DataProxyMetricItem();
-        this.metricItem.clusterId = CommonPropertiesHolder.getString(RemoteConfigManager.KEY_PROXY_CLUSTER_NAME);
+        this.metricItem.clusterId = metricName;
         final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
         StringBuilder beanName = new StringBuilder();
-        beanName.append(JMX_DOMAIN).append(DOMAIN_SEPARATOR).append("type=DataProxyCounter");
-        this.metricName = beanName.toString();
+        beanName.append(JMX_DOMAIN).append(DOMAIN_SEPARATOR).append("type=DataProxyPrometheus");
+        String strBeanName = beanName.toString();
         try {
-            ObjectName objName = new ObjectName(metricName);
+            ObjectName objName = new ObjectName(strBeanName);
             mbs.registerMBean(metricItem, objName);
         } catch (Exception ex) {
-            LOG.error("exception while register mbean:{},error:{}", metricName, ex.getMessage());
+            LOG.error("exception while register mbean:{},error:{}", strBeanName, ex);
         }
-        //
+        // prepare metric value map
         metricValueMap.put(M_READ_SUCCESS_COUNT, metricItem.readSuccessCount);
         metricValueMap.put(M_READ_SUCCESS_SIZE, metricItem.readSuccessSize);
         metricValueMap.put(M_READ_FAIL_COUNT, metricItem.readFailCount);
@@ -115,6 +115,7 @@ public PrometheusMetricListener() {
         int httpPort = CommonPropertiesHolder.getInteger(KEY_PROMETHEUS_HTTP_PORT, DEFAULT_PROMETHEUS_HTTP_PORT);
         try {
             this.httpServer = new HTTPServer(httpPort);
+            this.register();
         } catch (IOException e) {
             LOG.error("exception while register prometheus http server:{},error:{}", metricName, e.getMessage());
         }
@@ -142,7 +143,7 @@ public void snapshot(String domain, List<MetricItemValue> itemValues) {
             // id dimension
             String dimensionKey = itemValue.getKey();
             MetricItemValue dimensionMetricValue = this.dimensionMetricValueMap.get(dimensionKey);
-            if (dimensionKey == null) {
+            if (dimensionMetricValue == null) {
                 dimensionMetricValue = new MetricItemValue(dimensionKey, new ConcurrentHashMap<String, String>(),
                         new ConcurrentHashMap<String, MetricValue>());
                 this.dimensionMetricValueMap.putIfAbsent(dimensionKey, dimensionMetricValue);
@@ -202,7 +203,7 @@ public List<MetricFamilySamples> collect() {
 
         // id dimension
         CounterMetricFamily idCounter = new CounterMetricFamily(metricName + "&group=id",
-                "The metrics of inlong dataflow.", this.dimensionKeys);
+                "The metrics of inlong datastream.", this.dimensionKeys);
         for (Entry<String, MetricItemValue> entry : this.dimensionMetricValueMap.entrySet()) {
             MetricItemValue itemValue = entry.getValue();
             // read

inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/metrics/TestMetricListenerRunnable.java

@@ -26,7 +26,6 @@
 import org.apache.inlong.commons.config.metrics.MetricRegister;
 import org.apache.inlong.commons.config.metrics.MetricUtils;
 import org.apache.inlong.commons.config.metrics.MetricValue;
-import org.junit.BeforeClass;
 import org.junit.Test;
 
 /**
@@ -54,10 +53,12 @@ public class TestMetricListenerRunnable {
     private static String keySink2;
 
     /**
-     * setup
+     * testResult
+     * 
+     * @throws Exception
      */
-    @BeforeClass
-    public static void setup() {
+    @Test
+    public void testResult() throws Exception {
         itemSet = new DataProxyMetricItemSet(CLUSTER_ID);
         MetricRegister.register(itemSet);
         // prepare
@@ -76,15 +77,6 @@ public static void setup() {
         itemSink.inlongGroupId = INLONG_GROUP_ID1;
         itemSink.inlongStreamId = INLONG_STREAM_ID;
         dimSink = itemSink.getDimensions();
-    }
-
-    /**
-     * testResult
-     * 
-     * @throws Exception
-     */
-    @Test
-    public void testResult() throws Exception {
         // increase source
         DataProxyMetricItem item = null;
         item = itemSet.findMetricItem(dimSource);
@@ -146,6 +138,7 @@ public void snapshot(String domain, List<MetricItemValue> itemValues) {
         List<MetricListener> listeners = new ArrayList<>();
         listeners.add(listener);
         MetricListenerRunnable runnable = new MetricListenerRunnable("DataProxy", listeners);
+        List<MetricItemValue> itemValues = runnable.getItemValues();
         runnable.run();
     }
 }