You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by go...@apache.org on 2022/02/16 08:37:44 UTC
[incubator-inlong] branch master updated: [INLONG-2516][TubeMQ] Optimize Broker's JMX metric interface (#2519)
This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 3837074 [INLONG-2516][TubeMQ] Optimize Broker's JMX metric interface (#2519)
3837074 is described below
commit 3837074c923124d38ac241f3f24b044aa5854693
Author: gosonzhang <46...@qq.com>
AuthorDate: Wed Feb 16 16:37:38 2022 +0800
[INLONG-2516][TubeMQ] Optimize Broker's JMX metric interface (#2519)
---
.../tubemq/corebase/metric/MetricMXBean.java} | 18 +-
.../inlong/tubemq/server/broker/TubeBroker.java | 2 +
.../server/broker/metrics/BrokerMetrics.java | 124 -----------
.../server/broker/metrics/BrokerMetricsHolder.java | 88 --------
.../server/broker/stats/BrokerJMXHolder.java | 117 +++++++++++
.../tubemq/server/broker/BrokerMetricsTest.java | 230 ---------------------
6 files changed, 128 insertions(+), 451 deletions(-)
diff --git a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricMXBean.java b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricMXBean.java
similarity index 69%
rename from inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricMXBean.java
rename to inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricMXBean.java
index 3058556..16ad2e9 100644
--- a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricMXBean.java
+++ b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricMXBean.java
@@ -15,22 +15,22 @@
* limitations under the License.
*/
-package org.apache.inlong.tubemq.server.broker.metrics;
+package org.apache.inlong.tubemq.corebase.metric;
-import org.apache.inlong.tubemq.corebase.metric.MetricValues;
+import java.util.Map;
/**
- * BrokerMetricMXBean
+ * MetricMXBean
* Broker's metric data access interface, including:
- * the getMetric() that directly obtains data
- * the getAndReSetMetrics() that can clear the values of
+ * the getValue() that directly obtains data
+ * the snapshot() that can clear the values of
* the counter, maximum and minimum extremum Gauge data
*/
-public interface BrokerMetricMXBean {
+public interface MetricMXBean {
// get current metric data by viewing mode
- MetricValues getMetrics();
+ Map<String, Long> getValue();
- // get current metric data and reset the Counter, maximum/minimum Gauge metric
- MetricValues getAndReSetMetrics();
+ // get current metric data and reset the metric
+ Map<String, Long> snapshot();
}
diff --git a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
index 6b34a20..7257730 100644
--- a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
+++ b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
@@ -53,6 +53,7 @@ import org.apache.inlong.tubemq.server.broker.nodeinfo.ConsumerNodeInfo;
import org.apache.inlong.tubemq.server.broker.offset.DefaultOffsetManager;
import org.apache.inlong.tubemq.server.broker.offset.OffsetRecordService;
import org.apache.inlong.tubemq.server.broker.offset.OffsetService;
+import org.apache.inlong.tubemq.server.broker.stats.BrokerJMXHolder;
import org.apache.inlong.tubemq.server.broker.stats.ServiceStatsHolder;
import org.apache.inlong.tubemq.server.broker.utils.BrokerSamplePrint;
import org.apache.inlong.tubemq.server.broker.web.WebServer;
@@ -115,6 +116,7 @@ public class TubeBroker implements Stoppable {
java.security.Security.setProperty("networkaddress.cache.negative.ttl", "1");
this.tubeConfig = tubeConfig;
this.brokerId = generateBrokerClientId();
+ BrokerJMXHolder.registerMXBean();
this.metadataManager = new BrokerMetadataManager();
this.offsetManager = new DefaultOffsetManager(tubeConfig);
this.storeManager = new MessageStoreManager(this, tubeConfig);
diff --git a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetrics.java b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetrics.java
deleted file mode 100644
index f95d49d..0000000
--- a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetrics.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.tubemq.server.broker.metrics;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-import org.apache.inlong.tubemq.corebase.metric.AbsMetricItem;
-import org.apache.inlong.tubemq.corebase.metric.CountMetricItem;
-import org.apache.inlong.tubemq.corebase.metric.GaugeNormMetricItem;
-import org.apache.inlong.tubemq.corebase.metric.MetricValues;
-import org.apache.inlong.tubemq.corebase.metric.TimeDltMetricItem;
-import org.apache.inlong.tubemq.corebase.utils.DateTimeConvertUtils;
-
-public class BrokerMetrics implements BrokerMetricMXBean {
-
- private final AtomicLong lastResetTime =
- new AtomicLong(System.currentTimeMillis());
- // Delay statistics for syncing data to files
- protected final TimeDltMetricItem fileSyncTimeDltItem =
- new TimeDltMetricItem("file_sync");
- // Delay statistics for syncing data to Zookeeper
- protected final TimeDltMetricItem zkSyncTimeDltItem =
- new TimeDltMetricItem("zk_sync");
- // Zookeeper Exception statistics
- protected final AbsMetricItem zkExceptionCnt =
- new CountMetricItem("zk_exception_cnt");
- // Broker 2 Master status statistics
- protected final AbsMetricItem masterNoNodeCnt =
- new CountMetricItem("online_timeout_cnt");
- protected final AbsMetricItem hbExceptionCnt =
- new CountMetricItem("hb_master_exception_cnt");
- // Disk IO Exception statistics
- protected final AbsMetricItem ioExceptionCnt =
- new CountMetricItem("io_exception_cnt");
- // Consumer client statistics
- protected final AbsMetricItem consumerOnlineCnt =
- new GaugeNormMetricItem("consumer_online_cnt");
- protected final AbsMetricItem consumerTmoTotCnt =
- new CountMetricItem("consumer_timeout_cnt");
-
- @Override
- public MetricValues getMetrics() {
- return snapshotMetrics(false);
- }
-
- @Override
- public MetricValues getAndReSetMetrics() {
- return snapshotMetrics(true);
- }
-
- public long getLastResetTime() {
- return lastResetTime.get();
- }
-
- public AbsMetricItem getZkExceptionCnt() {
- return zkExceptionCnt;
- }
-
- public AbsMetricItem getMasterNoNodeCnt() {
- return masterNoNodeCnt;
- }
-
- public AbsMetricItem getHbExceptionCnt() {
- return hbExceptionCnt;
- }
-
- public AbsMetricItem getIoExceptionCnt() {
- return ioExceptionCnt;
- }
-
- public AbsMetricItem getConsumerOnlineCnt() {
- return consumerOnlineCnt;
- }
-
- public AbsMetricItem getConsumerTmoTotCnt() {
- return consumerTmoTotCnt;
- }
-
- public TimeDltMetricItem getFileSyncTimeDltItem() {
- return fileSyncTimeDltItem;
- }
-
- public TimeDltMetricItem getZkSyncTimeDltItem() {
- return zkSyncTimeDltItem;
- }
-
- private MetricValues snapshotMetrics(boolean resetValue) {
- Map<String, Long> metricValues = new LinkedHashMap<>();
- fileSyncTimeDltItem.getProcTimeDltDuration(metricValues, resetValue);
- zkSyncTimeDltItem.getProcTimeDltDuration(metricValues, resetValue);
- metricValues.put(zkExceptionCnt.getName(), zkExceptionCnt.getValue(resetValue));
- metricValues.put(masterNoNodeCnt.getName(), masterNoNodeCnt.getValue(resetValue));
- metricValues.put(hbExceptionCnt.getName(), hbExceptionCnt.getValue(resetValue));
- metricValues.put(ioExceptionCnt.getName(), ioExceptionCnt.getValue(resetValue));
- metricValues.put(consumerOnlineCnt.getName(), consumerOnlineCnt.getValue(resetValue));
- metricValues.put(consumerTmoTotCnt.getName(), consumerTmoTotCnt.getValue(resetValue));
- if (resetValue) {
- long befTime = lastResetTime.getAndSet(System.currentTimeMillis());
- return new MetricValues(
- DateTimeConvertUtils.ms2yyyyMMddHHmmss(befTime), metricValues);
- } else {
- return new MetricValues(
- DateTimeConvertUtils.ms2yyyyMMddHHmmss(lastResetTime.get()), metricValues);
- }
- }
-
-}
-
diff --git a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricsHolder.java b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricsHolder.java
deleted file mode 100644
index ef7cf81..0000000
--- a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metrics/BrokerMetricsHolder.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.tubemq.server.broker.metrics;
-
-import java.lang.management.ManagementFactory;
-import java.util.concurrent.atomic.AtomicBoolean;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class BrokerMetricsHolder {
- private static final Logger logger =
- LoggerFactory.getLogger(BrokerMetricsHolder.class);
- // Registration status indicator
- private static final AtomicBoolean registered = new AtomicBoolean(false);
- // broker metrics information
- private static final BrokerMetrics statsInfo = new BrokerMetrics();
-
- public static void registerMXBean() {
- if (!registered.compareAndSet(false, true)) {
- return;
- }
- try {
- MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
- ObjectName mxBeanName =
- new ObjectName("org.apache.inlong.tubemq.server.broker:type=BrokerMetrics");
- mbs.registerMBean(statsInfo, mxBeanName);
- } catch (Exception ex) {
- logger.error("Register BrokerMXBean error: ", ex);
- }
- }
-
- public static void incConsumerCnt() {
- statsInfo.consumerOnlineCnt.incrementAndGet();
- }
-
- public static void decConsumerCnt(boolean isTimeout) {
- statsInfo.consumerOnlineCnt.decrementAndGet();
- if (isTimeout) {
- statsInfo.consumerTmoTotCnt.incrementAndGet();
- }
- }
-
- public static void incMasterNoNodeCnt() {
- statsInfo.masterNoNodeCnt.incrementAndGet();
- }
-
- public static void incHBExceptionCnt() {
- statsInfo.hbExceptionCnt.incrementAndGet();
- }
-
- public static void incIOExceptionCnt() {
- statsInfo.ioExceptionCnt.incrementAndGet();
- }
-
- public static void incZKExceptionCnt() {
- statsInfo.zkExceptionCnt.incrementAndGet();
- }
-
- public static void updSyncDataDurations(long dltTime) {
- statsInfo.fileSyncTimeDltItem.updProcTimeDlt(dltTime);
- }
-
- public static void updSyncZKDurations(long dltTime) {
- statsInfo.zkSyncTimeDltItem.updProcTimeDlt(dltTime);
- }
-
- public static BrokerMetrics getStatsInfo() {
- return statsInfo;
- }
-}
-
diff --git a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/BrokerJMXHolder.java b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/BrokerJMXHolder.java
new file mode 100644
index 0000000..5fe4d79
--- /dev/null
+++ b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/BrokerJMXHolder.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.tubemq.server.broker.stats;
+
+import java.lang.management.ManagementFactory;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import org.apache.inlong.tubemq.corebase.metric.MetricMXBean;
+import org.apache.inlong.tubemq.server.common.webbase.WebCallStatsHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * BrokerJMXHolder
+ *
+ * A wrapper class for Broker JMX metric display, which currently includes RPC service status
+ * and web API call status metric data output
+ */
+public class BrokerJMXHolder {
+ private static final Logger logger =
+ LoggerFactory.getLogger(BrokerJMXHolder.class);
+ // Registration status indicator
+ private static final AtomicBoolean registered = new AtomicBoolean(false);
+ // broker metrics information
+ private static final BrokerServiceStatusBean serviceStatusInfo =
+ new BrokerServiceStatusBean();
+ // broker web api status information
+ private static final BrokerWebAPIStatusBean webAPIStatusInfo =
+ new BrokerWebAPIStatusBean();
+
+ /**
+ * Register MXBean
+ *
+ */
+ public static void registerMXBean() {
+ if (!registered.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ // register service status jmx
+ ObjectName srvStatusMxBeanName =
+ new ObjectName("org.apache.inlong.tubemq.server.broker:type=serviceStatus");
+ mbs.registerMBean(serviceStatusInfo, srvStatusMxBeanName);
+ // register web api status jmx
+ ObjectName webAPIMxBeanName =
+ new ObjectName("org.apache.inlong.tubemq.server.broker:type=webAPI");
+ mbs.registerMBean(webAPIStatusInfo, webAPIMxBeanName);
+
+ } catch (Exception ex) {
+ logger.error("Register Broker MXBean error: ", ex);
+ }
+ }
+
+ /**
+ * BrokerServiceStatusBean
+ *
+ * Broker service status metric wrapper class
+ */
+ private static class BrokerServiceStatusBean implements MetricMXBean {
+
+ @Override
+ public Map<String, Long> getValue() {
+ Map<String, Long> metricValues = new LinkedHashMap<>();
+ ServiceStatsHolder.getValue(metricValues);
+ return metricValues;
+ }
+
+ @Override
+ public Map<String, Long> snapshot() {
+ Map<String, Long> metricValues = new LinkedHashMap<>();
+ ServiceStatsHolder.snapShort(metricValues);
+ return metricValues;
+ }
+ }
+
+ /**
+ * BrokerWebAPIStatusBean
+ *
+ * Broker web api status metric wrapper class
+ */
+ private static class BrokerWebAPIStatusBean implements MetricMXBean {
+
+ @Override
+ public Map<String, Long> getValue() {
+ Map<String, Long> metricValues = new LinkedHashMap<>();
+ WebCallStatsHolder.getValue(metricValues);
+ return metricValues;
+ }
+
+ @Override
+ public Map<String, Long> snapshot() {
+ Map<String, Long> metricValues = new LinkedHashMap<>();
+ WebCallStatsHolder.snapShort(metricValues);
+ return metricValues;
+ }
+ }
+}
+
diff --git a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/BrokerMetricsTest.java b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/BrokerMetricsTest.java
deleted file mode 100644
index a5d8e74..0000000
--- a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/BrokerMetricsTest.java
+++ /dev/null
@@ -1,230 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.tubemq.server.broker;
-
-import org.apache.inlong.tubemq.corebase.metric.MetricValues;
-import org.apache.inlong.tubemq.server.broker.metrics.BrokerMetrics;
-import org.apache.inlong.tubemq.server.broker.metrics.BrokerMetricsHolder;
-import org.junit.Assert;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class BrokerMetricsTest {
- private static final Logger logger =
- LoggerFactory.getLogger(BrokerMetricsTest.class);
-
- @Test
- public void testBrokerMetrics() {
- try {
- BrokerMetrics metrics = new BrokerMetrics();
- // test case 1, set data
- metrics.getIoExceptionCnt().incrementAndGet();
- metrics.getZkExceptionCnt().incrementAndGet();
- metrics.getConsumerOnlineCnt().incrementAndGet();
- metrics.getConsumerOnlineCnt().incrementAndGet();
- metrics.getConsumerTmoTotCnt().incrementAndGet();
- metrics.getHbExceptionCnt().incrementAndGet();
- metrics.getMasterNoNodeCnt().incrementAndGet();
-
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(20000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(10000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(30000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(2000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(1000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(3000);
-
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(20000);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(1000);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(30000);
-
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(2000);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(100);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(3000);
- // get metric and compare data
- MetricValues result1 = metrics.getMetrics();
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getIoExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getZkExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(2),
- result1.getMetricValues().get(metrics.getConsumerOnlineCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getConsumerTmoTotCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getHbExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getMasterNoNodeCnt().getName()));
- Assert.assertEquals(Long.valueOf(30000), result1.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(1000), result1.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMinJmxKey()));
- Assert.assertEquals(Long.valueOf(30000), result1.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(100), result1.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMinJmxKey()));
- // get and reset value 2
- final MetricValues result2 = metrics.getAndReSetMetrics();
- // update metric data to 3
- metrics.getIoExceptionCnt().incrementAndGet();
- metrics.getZkExceptionCnt().incrementAndGet();
- metrics.getConsumerOnlineCnt().incrementAndGet();
- metrics.getConsumerOnlineCnt().decrementAndGet();
- metrics.getConsumerTmoTotCnt().incrementAndGet();
- metrics.getHbExceptionCnt().incrementAndGet();
- metrics.getMasterNoNodeCnt().incrementAndGet();
-
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(10);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(10000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(20000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(10);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(1000);
- metrics.getFileSyncTimeDltItem().updProcTimeDlt(5000);
-
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(10);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(1000);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(2000);
-
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(3000);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(10);
- metrics.getZkSyncTimeDltItem().updProcTimeDlt(6000);
-
- MetricValues result3 = metrics.getMetrics();
- Assert.assertEquals(result1.getLastResetTime(),
- result2.getLastResetTime());
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getIoExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getZkExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(2),
- result3.getMetricValues().get(metrics.getConsumerOnlineCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getConsumerTmoTotCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getHbExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getMasterNoNodeCnt().getName()));
- Assert.assertEquals(Long.valueOf(20000), result3.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(10), result3.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMinJmxKey()));
- Assert.assertEquals(Long.valueOf(20000), result3.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(10), result3.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMinJmxKey()));
- } catch (Exception ex) {
- logger.error("error happens" + ex);
- }
- }
-
- @Test
- public void testBrokerMetricsHolder() {
- try {
- // case 1, set data
- BrokerMetricsHolder.incConsumerCnt();
- BrokerMetricsHolder.decConsumerCnt(false);
- BrokerMetricsHolder.incConsumerCnt();
- BrokerMetricsHolder.decConsumerCnt(true);
- BrokerMetricsHolder.incConsumerCnt();
-
- BrokerMetricsHolder.incZKExceptionCnt();
- BrokerMetricsHolder.incZKExceptionCnt();
-
- BrokerMetricsHolder.incMasterNoNodeCnt();
- BrokerMetricsHolder.incHBExceptionCnt();
- BrokerMetricsHolder.incIOExceptionCnt();
- BrokerMetricsHolder.incZKExceptionCnt();
-
- BrokerMetricsHolder.updSyncDataDurations(10000);
- BrokerMetricsHolder.updSyncDataDurations(2000);
- BrokerMetricsHolder.updSyncDataDurations(20000);
-
- BrokerMetricsHolder.updSyncZKDurations(1000);
- BrokerMetricsHolder.updSyncZKDurations(30);
- BrokerMetricsHolder.updSyncZKDurations(30000);
- // get data and check
- BrokerMetrics metrics = BrokerMetricsHolder.getStatsInfo();
- MetricValues result1 = metrics.getMetrics();
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getConsumerOnlineCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getConsumerTmoTotCnt().getName()));
- Assert.assertEquals(Long.valueOf(3),
- result1.getMetricValues().get(metrics.getZkExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getMasterNoNodeCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getHbExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result1.getMetricValues().get(metrics.getIoExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(20000), result1.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(2000), result1.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMinJmxKey()));
- Assert.assertEquals(Long.valueOf(30000), result1.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(30), result1.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMinJmxKey()));
-
- // get and reset value 2
- final MetricValues result2 = metrics.getAndReSetMetrics();
- BrokerMetricsHolder.incConsumerCnt();
- BrokerMetricsHolder.incConsumerCnt();
- BrokerMetricsHolder.incConsumerCnt();
- BrokerMetricsHolder.decConsumerCnt(false);
- BrokerMetricsHolder.decConsumerCnt(true);
-
- BrokerMetricsHolder.incZKExceptionCnt();
- BrokerMetricsHolder.incZKExceptionCnt();
-
- BrokerMetricsHolder.updSyncDataDurations(1);
- BrokerMetricsHolder.updSyncDataDurations(5000);
- BrokerMetricsHolder.updSyncDataDurations(30000);
-
- BrokerMetricsHolder.updSyncZKDurations(100);
- BrokerMetricsHolder.updSyncZKDurations(10);
- BrokerMetricsHolder.updSyncZKDurations(5000);
- // get and check 3
- MetricValues result3 = metrics.getMetrics();
- Assert.assertEquals(result1.getLastResetTime(),
- result2.getLastResetTime());
- Assert.assertEquals(Long.valueOf(2),
- result3.getMetricValues().get(metrics.getConsumerOnlineCnt().getName()));
- Assert.assertEquals(Long.valueOf(1),
- result3.getMetricValues().get(metrics.getConsumerTmoTotCnt().getName()));
- Assert.assertEquals(Long.valueOf(2),
- result3.getMetricValues().get(metrics.getZkExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(0),
- result3.getMetricValues().get(metrics.getMasterNoNodeCnt().getName()));
- Assert.assertEquals(Long.valueOf(0),
- result3.getMetricValues().get(metrics.getHbExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(0),
- result3.getMetricValues().get(metrics.getIoExceptionCnt().getName()));
- Assert.assertEquals(Long.valueOf(30000), result3.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(1), result3.getMetricValues().get(
- metrics.getFileSyncTimeDltItem().getDltMinJmxKey()));
- Assert.assertEquals(Long.valueOf(5000), result3.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMaxJmxKey()));
- Assert.assertEquals(Long.valueOf(10), result3.getMetricValues().get(
- metrics.getZkSyncTimeDltItem().getDltMinJmxKey()));
- } catch (Exception ex) {
- logger.error("error happens" + ex);
- }
- }
-}