You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2022/12/01 05:08:57 UTC
[james-project] 04/08: [JAMES-3841] ActiveMQ: add SettableGauge to metrics api and use it in ActiveMQ metric collection
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 6ae8df410641c34d0851394fde08ec8cefdd6bb7
Author: ouvtam <ou...@8n4.pw>
AuthorDate: Tue Nov 29 01:02:05 2022 +0100
[JAMES-3841] ActiveMQ: add SettableGauge to metrics api and use it in ActiveMQ metric collection
---
.../apache/james/metrics/api/GaugeRegistry.java | 6 +
.../james/metrics/api/NoopGaugeRegistry.java | 5 +
.../dropwizard/DropWizardGaugeRegistry.java | 8 +
.../activemq/metric/ActiveMQBrokerStatistics.java | 106 -------------
.../metric/ActiveMQMetricCollectorImpl.java | 13 +-
.../queue/activemq/metric/ActiveMQMetrics.java | 176 +++++++++++++++++++++
.../activemq/metric/ActiveMQQueueStatistics.java | 155 ------------------
.../queue/activemq/metric/ActiveMQStatistics.java | 34 ----
.../metric/ActiveMQMetricCollectorTest.java | 43 +++--
9 files changed, 227 insertions(+), 319 deletions(-)
diff --git a/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/GaugeRegistry.java b/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/GaugeRegistry.java
index d35062749f..6d2c6d3c21 100644
--- a/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/GaugeRegistry.java
+++ b/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/GaugeRegistry.java
@@ -20,5 +20,11 @@
package org.apache.james.metrics.api;
public interface GaugeRegistry {
+ interface SettableGauge<T> {
+ void setValue(T t);
+ }
+
<T> GaugeRegistry register(String name, Gauge<T> gauge);
+
+ <T> SettableGauge<T> settableGauge(String name);
}
diff --git a/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/NoopGaugeRegistry.java b/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/NoopGaugeRegistry.java
index 80fe43badc..07cb98b167 100644
--- a/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/NoopGaugeRegistry.java
+++ b/metrics/metrics-api/src/main/java/org/apache/james/metrics/api/NoopGaugeRegistry.java
@@ -24,4 +24,9 @@ public class NoopGaugeRegistry implements GaugeRegistry {
public <T> GaugeRegistry register(String name, Gauge<T> gauge) {
return this;
}
+
+ @Override
+ public <T> SettableGauge<T> settableGauge(String name) {
+ return t -> { };
+ }
}
diff --git a/metrics/metrics-dropwizard/src/main/java/org/apache/james/metrics/dropwizard/DropWizardGaugeRegistry.java b/metrics/metrics-dropwizard/src/main/java/org/apache/james/metrics/dropwizard/DropWizardGaugeRegistry.java
index f79e2832ee..7f858a4818 100644
--- a/metrics/metrics-dropwizard/src/main/java/org/apache/james/metrics/dropwizard/DropWizardGaugeRegistry.java
+++ b/metrics/metrics-dropwizard/src/main/java/org/apache/james/metrics/dropwizard/DropWizardGaugeRegistry.java
@@ -25,6 +25,7 @@ import javax.inject.Inject;
import org.apache.james.metrics.api.Gauge;
import org.apache.james.metrics.api.GaugeRegistry;
+import com.codahale.metrics.DefaultSettableGauge;
import com.codahale.metrics.MetricRegistry;
public class DropWizardGaugeRegistry implements GaugeRegistry {
@@ -45,4 +46,11 @@ public class DropWizardGaugeRegistry implements GaugeRegistry {
public void shutDown() {
metricRegistry.getGauges().keySet().forEach(metricRegistry::remove);
}
+
+ @Override
+ public <T> SettableGauge<T> settableGauge(String name) {
+ DefaultSettableGauge<T> gauge = new DefaultSettableGauge<>();
+ metricRegistry.register(name, gauge);
+ return gauge::setValue;
+ }
}
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQBrokerStatistics.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQBrokerStatistics.java
deleted file mode 100644
index e7ae8acbd3..0000000000
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQBrokerStatistics.java
+++ /dev/null
@@ -1,106 +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.james.queue.activemq.metric;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-
-import org.apache.james.metrics.api.GaugeRegistry;
-
-public class ActiveMQBrokerStatistics extends ActiveMQQueueStatistics {
- public static final String TEMP_PERCENT_USAGE = "tempPercentUsage";
- public static final String STORE_PERCENT_USAGE = "storePercentUsage";
- public static final String MEMORY_PERCENT_USAGE = "memoryPercentUsage";
-
- public static final String STORE_LIMIT = "storeLimit";
- public static final String TEMP_LIMIT = "tempLimit";
-
- public static final String MEMORY_USAGE = "memoryUsage";
- public static final String STORE_USAGE = "storeUsage";
- public static final String TEMP_USAGE = "tempUsage";
-
- private final AtomicInteger memoryPercentUsage = new AtomicInteger();
- private final AtomicInteger storePercentUsage = new AtomicInteger();
- private final AtomicInteger tempPercentUsage = new AtomicInteger();
-
- private final AtomicLong storeLimit = new AtomicLong();
- private final AtomicLong tempLimit = new AtomicLong();
-
- private final AtomicLong memoryUsage = new AtomicLong();
- private final AtomicLong storeUsage = new AtomicLong();
- private final AtomicLong tempUsage = new AtomicLong();
-
- public ActiveMQBrokerStatistics() {
- super("ActiveMQ.Statistics.Broker");
- }
-
- @Override
- public void registerMetrics(GaugeRegistry gaugeRegistry) {
- super.registerMetrics(gaugeRegistry);
-
- String prefix = getName() + ".";
- gaugeRegistry.register(prefix + MEMORY_PERCENT_USAGE, memoryPercentUsage::get);
- gaugeRegistry.register(prefix + STORE_PERCENT_USAGE, storePercentUsage::get);
- gaugeRegistry.register(prefix + TEMP_PERCENT_USAGE, tempPercentUsage::get);
-
- gaugeRegistry.register(prefix + STORE_LIMIT, storeLimit::get);
- gaugeRegistry.register(prefix + TEMP_LIMIT, tempLimit::get);
-
- gaugeRegistry.register(prefix + MEMORY_USAGE, memoryUsage::get);
- gaugeRegistry.register(prefix + STORE_USAGE, storeUsage::get);
- gaugeRegistry.register(prefix + TEMP_USAGE, tempUsage::get);
- }
-
- @Override
- public void update(MapMessage msg) throws JMSException {
- super.update(msg);
-
- if (msg.itemExists(MEMORY_PERCENT_USAGE)) {
- memoryPercentUsage.set(msg.getInt(MEMORY_PERCENT_USAGE));
- }
- if (msg.itemExists(STORE_PERCENT_USAGE)) {
- storePercentUsage.set(msg.getInt(STORE_PERCENT_USAGE));
- }
- if (msg.itemExists(TEMP_PERCENT_USAGE)) {
- tempPercentUsage.set(msg.getInt(TEMP_PERCENT_USAGE));
- }
-
- if (msg.itemExists(STORE_LIMIT)) {
- storeLimit.set(msg.getLong(STORE_LIMIT));
- }
- if (msg.itemExists(TEMP_LIMIT)) {
- tempLimit.set(msg.getLong(TEMP_LIMIT));
- }
-
- if (msg.itemExists(MEMORY_USAGE)) {
- memoryUsage.set(msg.getLong(MEMORY_USAGE));
- }
- if (msg.itemExists(STORE_USAGE)) {
- storeUsage.set(msg.getLong(STORE_USAGE));
- }
- if (msg.itemExists(TEMP_USAGE)) {
- tempUsage.set(msg.getLong(TEMP_USAGE));
- }
- }
-
-}
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorImpl.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorImpl.java
index 1585969abf..d4ed4856b0 100644
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorImpl.java
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorImpl.java
@@ -62,7 +62,7 @@ public class ActiveMQMetricCollectorImpl implements ActiveMQMetricCollector {
private final MetricFactory metricFactory;
private final GaugeRegistry gaugeRegistry;
- private final Map<String, ActiveMQStatistics> registeredStatistics = new HashMap<>();
+ private final Map<String, ActiveMQMetrics> registeredStatistics = new HashMap<>();
private Disposable disposable;
@@ -75,19 +75,18 @@ public class ActiveMQMetricCollectorImpl implements ActiveMQMetricCollector {
@Override
public void collectBrokerStatistics() {
- collectStatistics(new ActiveMQBrokerStatistics());
+ collectStatistics(ActiveMQMetrics.forBroker(gaugeRegistry));
}
@Override
public void collectQueueStatistics(MailQueueName name) {
- collectStatistics(ActiveMQQueueStatistics.from(name.asString()));
+ collectStatistics(ActiveMQMetrics.forQueue(name.asString(), gaugeRegistry));
}
- private void collectStatistics(ActiveMQStatistics statistics) {
+ private void collectStatistics(ActiveMQMetrics statistics) {
if (!registeredStatistics.containsKey(statistics.getName())) {
LOGGER.info("collecting statistics for {}", statistics.getName());
registeredStatistics.put(statistics.getName(), statistics);
- statistics.registerMetrics(gaugeRegistry);
}
}
@@ -120,7 +119,7 @@ public class ActiveMQMetricCollectorImpl implements ActiveMQMetricCollector {
}
@VisibleForTesting
- Void fetchAndUpdate(ActiveMQStatistics stats) throws JMSException {
+ Void fetchAndUpdate(ActiveMQMetrics stats) throws JMSException {
Connection connection = null;
Session session = null;
TemporaryQueue replyTo = null;
@@ -146,7 +145,7 @@ public class ActiveMQMetricCollectorImpl implements ActiveMQMetricCollector {
} else if (!(reply instanceof MapMessage)) {
throw new JMSException("expected MapMessage but got " + reply.getClass());
}
- stats.update((MapMessage)reply);
+ stats.updateMetrics((MapMessage)reply);
} finally {
if (producer != null) {
producer.close();
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetrics.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetrics.java
new file mode 100644
index 0000000000..6a6a1bfc37
--- /dev/null
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQMetrics.java
@@ -0,0 +1,176 @@
+/****************************************************************
+ * 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.james.queue.activemq.metric;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+
+import org.apache.james.metrics.api.GaugeRegistry;
+import org.apache.james.metrics.api.GaugeRegistry.SettableGauge;
+
+public class ActiveMQMetrics {
+ private static final String TEMP_PERCENT_USAGE = "tempPercentUsage";
+ private static final String STORE_PERCENT_USAGE = "storePercentUsage";
+ private static final String MEMORY_PERCENT_USAGE = "memoryPercentUsage";
+
+ private static final String MEMORY_LIMIT = "memoryLimit";
+ private static final String STORE_LIMIT = "storeLimit";
+ private static final String TEMP_LIMIT = "tempLimit";
+
+ private static final String MEMORY_USAGE = "memoryUsage";
+ private static final String STORE_USAGE = "storeUsage";
+ private static final String TEMP_USAGE = "tempUsage";
+
+ private static final String SIZE = "size";
+ private static final String ENQUEUE_COUNT = "enqueueCount";
+ private static final String DEQUEUE_COUNT = "dequeueCount";
+ private static final String INFLIGHT_COUNT = "inflightCount";
+ private static final String PRODUCER_COUNT = "producerCount";
+ private static final String CONSUMER_COUNT = "consumerCount";
+ private static final String EXPIRED_COUNT = "expiredCount";
+ private static final String DISPATCH_COUNT = "dispatchCount";
+ private static final String MESSAGES_CACHED = "messagesCached";
+
+ private static final String MIN_ENQUEUE_TIME = "minEnqueueTime";
+ private static final String MAX_ENQUEUE_TIME = "maxEnqueueTime";
+ private static final String AVERAGE_ENQUEUE_TIME = "averageEnqueueTime";
+
+ private static final String LAST_UPDATE = "lastUpdate";
+
+ private final String statsQueueName;
+
+ private final GaugeRegistry gaugeRegistry;
+
+ private final Map<String, SettableGauge<Integer>> registeredGaugesOfInteger = new HashMap<>();
+ private final Map<String, SettableGauge<Double>> registeredGaugesOfDouble = new HashMap<>();
+ private final Map<String, SettableGauge<Long>> registeredGaugesOfLong = new HashMap<>();
+
+ public static ActiveMQMetrics forQueue(String queueName, GaugeRegistry gaugeRegistry) {
+ return new ActiveMQMetrics("ActiveMQ.Statistics.Destination." + queueName, gaugeRegistry);
+ }
+
+ public static ActiveMQMetrics forBroker(GaugeRegistry gaugeRegistry) {
+ return new ActiveMQMetrics("ActiveMQ.Statistics.Broker", gaugeRegistry);
+ }
+
+ private ActiveMQMetrics(String statsQueueName, GaugeRegistry gaugeRegistry) {
+ this.statsQueueName = statsQueueName;
+ this.gaugeRegistry = gaugeRegistry;
+ }
+
+ public String getName() {
+ return statsQueueName;
+ }
+
+ public void updateMetrics(MapMessage msg) throws JMSException {
+
+ if (msg.itemExists(MEMORY_PERCENT_USAGE)) {
+ setGaugeAndRegisterIfAbsent(MEMORY_PERCENT_USAGE, msg.getInt(MEMORY_PERCENT_USAGE));
+ }
+ if (msg.itemExists(STORE_PERCENT_USAGE)) {
+ setGaugeAndRegisterIfAbsent(TEMP_PERCENT_USAGE, msg.getInt(STORE_PERCENT_USAGE));
+ }
+ if (msg.itemExists(TEMP_PERCENT_USAGE)) {
+ setGaugeAndRegisterIfAbsent(TEMP_PERCENT_USAGE, msg.getInt(TEMP_PERCENT_USAGE));
+ }
+
+ if (msg.itemExists(MEMORY_LIMIT)) {
+ setGaugeAndRegisterIfAbsent(MEMORY_LIMIT, msg.getLong(MEMORY_LIMIT));
+ }
+ if (msg.itemExists(STORE_LIMIT)) {
+ setGaugeAndRegisterIfAbsent(STORE_LIMIT, msg.getLong(STORE_LIMIT));
+ }
+ if (msg.itemExists(TEMP_LIMIT)) {
+ setGaugeAndRegisterIfAbsent(TEMP_LIMIT, msg.getLong(TEMP_LIMIT));
+ }
+
+ if (msg.itemExists(MEMORY_USAGE)) {
+ setGaugeAndRegisterIfAbsent(MEMORY_USAGE, msg.getLong(MEMORY_USAGE));
+ }
+ if (msg.itemExists(STORE_USAGE)) {
+ setGaugeAndRegisterIfAbsent(STORE_USAGE, msg.getLong(STORE_USAGE));
+ }
+ if (msg.itemExists(TEMP_USAGE)) {
+ setGaugeAndRegisterIfAbsent(TEMP_USAGE, msg.getLong(TEMP_USAGE));
+ }
+
+ if (msg.itemExists(SIZE)) {
+ setGaugeAndRegisterIfAbsent(SIZE, msg.getLong(SIZE));
+ }
+ if (msg.itemExists(ENQUEUE_COUNT)) {
+ setGaugeAndRegisterIfAbsent(ENQUEUE_COUNT, msg.getLong(ENQUEUE_COUNT));
+ }
+ if (msg.itemExists(DEQUEUE_COUNT)) {
+ setGaugeAndRegisterIfAbsent(DEQUEUE_COUNT, msg.getLong(DEQUEUE_COUNT));
+ }
+ if (msg.itemExists(INFLIGHT_COUNT)) {
+ setGaugeAndRegisterIfAbsent(INFLIGHT_COUNT, msg.getLong(INFLIGHT_COUNT));
+ }
+ if (msg.itemExists(PRODUCER_COUNT)) {
+ setGaugeAndRegisterIfAbsent(PRODUCER_COUNT, msg.getLong(PRODUCER_COUNT));
+ }
+ if (msg.itemExists(CONSUMER_COUNT)) {
+ setGaugeAndRegisterIfAbsent(CONSUMER_COUNT, msg.getLong(CONSUMER_COUNT));
+ }
+ if (msg.itemExists(EXPIRED_COUNT)) {
+ setGaugeAndRegisterIfAbsent(EXPIRED_COUNT, msg.getLong(EXPIRED_COUNT));
+ }
+ if (msg.itemExists(DISPATCH_COUNT)) {
+ setGaugeAndRegisterIfAbsent(DISPATCH_COUNT, msg.getLong(DISPATCH_COUNT));
+ }
+ if (msg.itemExists(MESSAGES_CACHED)) {
+ setGaugeAndRegisterIfAbsent(MESSAGES_CACHED, msg.getLong(MESSAGES_CACHED));
+ }
+
+ if (msg.itemExists(MIN_ENQUEUE_TIME)) {
+ setGaugeAndRegisterIfAbsent(MIN_ENQUEUE_TIME, msg.getDouble(MIN_ENQUEUE_TIME));
+ }
+ if (msg.itemExists(MAX_ENQUEUE_TIME)) {
+ setGaugeAndRegisterIfAbsent(MAX_ENQUEUE_TIME, msg.getDouble(MAX_ENQUEUE_TIME));
+ }
+ if (msg.itemExists(AVERAGE_ENQUEUE_TIME)) {
+ setGaugeAndRegisterIfAbsent(AVERAGE_ENQUEUE_TIME, msg.getDouble(AVERAGE_ENQUEUE_TIME));
+ }
+
+ setGaugeAndRegisterIfAbsent(LAST_UPDATE, System.currentTimeMillis());
+ }
+
+ private void setGaugeAndRegisterIfAbsent(String name, long val) {
+ String key = statsQueueName + "." + name;
+ registeredGaugesOfLong.computeIfAbsent(key, any -> gaugeRegistry.settableGauge(key))
+ .setValue(val);
+ }
+
+ private void setGaugeAndRegisterIfAbsent(String name, int val) {
+ String key = statsQueueName + "." + name;
+ registeredGaugesOfInteger.computeIfAbsent(key, any -> gaugeRegistry.settableGauge(key))
+ .setValue(val);
+ }
+
+ private void setGaugeAndRegisterIfAbsent(String name, double val) {
+ String key = statsQueueName + "." + name;
+ registeredGaugesOfDouble.computeIfAbsent(key, any -> gaugeRegistry.settableGauge(key))
+ .setValue(val);
+ }
+
+}
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQQueueStatistics.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQQueueStatistics.java
deleted file mode 100644
index f92c99abb0..0000000000
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQQueueStatistics.java
+++ /dev/null
@@ -1,155 +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.james.queue.activemq.metric;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-
-import org.apache.james.metrics.api.GaugeRegistry;
-
-import com.google.common.util.concurrent.AtomicDouble;
-
-public class ActiveMQQueueStatistics implements ActiveMQStatistics {
- protected static final String MEMORY_LIMIT = "memoryLimit";
-
- protected static final String SIZE = "size";
- protected static final String ENQUEUE_COUNT = "enqueueCount";
- protected static final String DEQUEUE_COUNT = "dequeueCount";
- protected static final String INFLIGHT_COUNT = "inflightCount";
- protected static final String PRODUCER_COUNT = "producerCount";
- protected static final String CONSUMER_COUNT = "consumerCount";
- protected static final String EXPIRED_COUNT = "expiredCount";
- protected static final String DISPATCH_COUNT = "dispatchCount";
- protected static final String MESSAGES_CACHED = "messagesCached";
-
- protected static final String MIN_ENQUEUE_TIME = "minEnqueueTime";
- protected static final String MAX_ENQUEUE_TIME = "maxEnqueueTime";
- protected static final String AVERAGE_ENQUEUE_TIME = "averageEnqueueTime";
-
- protected static final String LAST_UPDATE = "lastUpdate";
-
- private final AtomicLong memoryLimit = new AtomicLong();
-
- private final AtomicLong size = new AtomicLong();
- private final AtomicLong enqueueCount = new AtomicLong();
- private final AtomicLong dequeueCount = new AtomicLong();
- private final AtomicLong inflightCount = new AtomicLong();
- private final AtomicLong producerCount = new AtomicLong();
- private final AtomicLong consumerCount = new AtomicLong();
- private final AtomicLong expiredCount = new AtomicLong();
- private final AtomicLong dispatchCount = new AtomicLong();
- private final AtomicLong messagesCached = new AtomicLong();
-
- private final AtomicDouble minEnqueueTime = new AtomicDouble();
- private final AtomicDouble maxEnqueueTime = new AtomicDouble();
- private final AtomicDouble averageEnqueueTime = new AtomicDouble();
-
- private final AtomicLong lastUpdate = new AtomicLong();
-
- protected final String statsQueueName;
-
- public static ActiveMQQueueStatistics from(String queueName) {
- return new ActiveMQQueueStatistics("ActiveMQ.Statistics.Destination." + queueName);
- }
-
- public ActiveMQQueueStatistics(String statsQueueName) {
- this.statsQueueName = statsQueueName;
- }
-
- @Override
- public String getName() {
- return statsQueueName;
- }
-
- @Override
- public void registerMetrics(GaugeRegistry gaugeRegistry) {
- String prefix = statsQueueName + ".";
-
- gaugeRegistry.register(prefix + MEMORY_LIMIT, memoryLimit::get);
-
- gaugeRegistry.register(prefix + SIZE, size::get);
- gaugeRegistry.register(prefix + ENQUEUE_COUNT, enqueueCount::get);
- gaugeRegistry.register(prefix + DEQUEUE_COUNT, dequeueCount::get);
- gaugeRegistry.register(prefix + INFLIGHT_COUNT, inflightCount::get);
- gaugeRegistry.register(prefix + PRODUCER_COUNT, producerCount::get);
- gaugeRegistry.register(prefix + CONSUMER_COUNT, consumerCount::get);
- gaugeRegistry.register(prefix + EXPIRED_COUNT, expiredCount::get);
- gaugeRegistry.register(prefix + DISPATCH_COUNT, dispatchCount::get);
- gaugeRegistry.register(prefix + MESSAGES_CACHED, messagesCached::get);
-
- gaugeRegistry.register(prefix + MIN_ENQUEUE_TIME, minEnqueueTime::get);
- gaugeRegistry.register(prefix + MAX_ENQUEUE_TIME, maxEnqueueTime::get);
- gaugeRegistry.register(prefix + AVERAGE_ENQUEUE_TIME, averageEnqueueTime::get);
-
- gaugeRegistry.register(prefix + LAST_UPDATE, lastUpdate::get);
- }
-
- @Override
- public void update(MapMessage msg) throws JMSException {
- if (msg.itemExists(MEMORY_LIMIT)) {
- memoryLimit.set(msg.getLong(MEMORY_LIMIT));
- }
- if (msg.itemExists(SIZE)) {
- size.set(msg.getLong(SIZE));
- }
- if (msg.itemExists(ENQUEUE_COUNT)) {
- enqueueCount.set(msg.getLong(ENQUEUE_COUNT));
- }
- if (msg.itemExists(DEQUEUE_COUNT)) {
- dequeueCount.set(msg.getLong(DEQUEUE_COUNT));
- }
- if (msg.itemExists(INFLIGHT_COUNT)) {
- inflightCount.set(msg.getLong(INFLIGHT_COUNT));
- }
- if (msg.itemExists(PRODUCER_COUNT)) {
- producerCount.set(msg.getLong(PRODUCER_COUNT));
- }
- if (msg.itemExists(CONSUMER_COUNT)) {
- consumerCount.set(msg.getLong(CONSUMER_COUNT));
- }
- if (msg.itemExists(EXPIRED_COUNT)) {
- expiredCount.set(msg.getLong(EXPIRED_COUNT));
- }
- if (msg.itemExists(DISPATCH_COUNT)) {
- dispatchCount.set(msg.getLong(DISPATCH_COUNT));
- }
- if (msg.itemExists(MESSAGES_CACHED)) {
- messagesCached.set(msg.getLong(MESSAGES_CACHED));
- }
-
- if (msg.itemExists(MIN_ENQUEUE_TIME)) {
- minEnqueueTime.set(msg.getDouble(MIN_ENQUEUE_TIME));
- }
- if (msg.itemExists(MAX_ENQUEUE_TIME)) {
- maxEnqueueTime.set(msg.getDouble(MAX_ENQUEUE_TIME));
- }
- if (msg.itemExists(AVERAGE_ENQUEUE_TIME)) {
- averageEnqueueTime.set(msg.getDouble(AVERAGE_ENQUEUE_TIME));
- }
-
- lastUpdate.set(System.currentTimeMillis());
- }
-
- public long getLastUpdate() {
- return lastUpdate.get();
- }
-}
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQStatistics.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQStatistics.java
deleted file mode 100644
index 52d10504e9..0000000000
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/metric/ActiveMQStatistics.java
+++ /dev/null
@@ -1,34 +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.james.queue.activemq.metric;
-
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-
-import org.apache.james.metrics.api.GaugeRegistry;
-
-public interface ActiveMQStatistics {
-
- String getName();
-
- void registerMetrics(GaugeRegistry gaugeRegistry);
-
- void update(MapMessage msg) throws JMSException;
-}
diff --git a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorTest.java b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorTest.java
index f8a00e89cf..d1d02dea2e 100644
--- a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorTest.java
+++ b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/metric/ActiveMQMetricCollectorTest.java
@@ -49,11 +49,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
class ActiveMQMetricCollectorTest {
private static ActiveMQConnectionFactory connectionFactory;
- //private BrokerService broker;
@BeforeAll
static void setup(BrokerService broker) {
- //this.broker = broker;
connectionFactory = new ActiveMQConnectionFactory("vm://localhost?create=false");
ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
prefetchPolicy.setQueuePrefetch(0);
@@ -62,38 +60,40 @@ class ActiveMQMetricCollectorTest {
@Test
void shouldFailToFetchAndUpdateStatisticsForUnknownQueue() {
- ActiveMQMetricCollectorImpl testee = new ActiveMQMetricCollectorImpl(connectionFactory, new RecordingMetricFactory(), new NoopGaugeRegistry());
- ActiveMQQueueStatistics queueStatistics = new ActiveMQQueueStatistics("UNKNOWN");
+ SimpleGaugeRegistry gaugeRegistry = new SimpleGaugeRegistry();
+ ActiveMQMetricCollectorImpl testee = new ActiveMQMetricCollectorImpl(connectionFactory, new RecordingMetricFactory(), gaugeRegistry);
+ ActiveMQMetrics queueStatistics = ActiveMQMetrics.forQueue("UNKNOWN", gaugeRegistry);
assertThatThrownBy(() -> testee.fetchAndUpdate(queueStatistics))
.isInstanceOf(JMSException.class);
- assertThat(queueStatistics.getLastUpdate())
- .isEqualTo(0);
+
+ assertThat(gaugeRegistry.getGauge("ActiveMQ.Statistics.Destination.UNKNOWN")).isNull();
}
@Test
void shouldFetchAndUpdateBrokerStatistics() throws Exception {
- ActiveMQMetricCollectorImpl testee = new ActiveMQMetricCollectorImpl(connectionFactory, new RecordingMetricFactory(), new NoopGaugeRegistry());
- ActiveMQBrokerStatistics brokerStatistics = new ActiveMQBrokerStatistics();
+ SimpleGaugeRegistry gaugeRegistry = new SimpleGaugeRegistry();
+ ActiveMQMetricCollectorImpl testee = new ActiveMQMetricCollectorImpl(connectionFactory, new RecordingMetricFactory(), gaugeRegistry);
+ ActiveMQMetrics brokerStatistics = ActiveMQMetrics.forBroker(gaugeRegistry);
long notBefore = System.currentTimeMillis();
testee.fetchAndUpdate(brokerStatistics);
- assertThat(brokerStatistics.getLastUpdate())
- .isGreaterThanOrEqualTo(notBefore);
+ Number n = gaugeRegistry.getGauge("ActiveMQ.Statistics.Broker.lastUpdate");
+ assertThat(n).isInstanceOf(Long.class);
+ assertThat((Long) n).isGreaterThanOrEqualTo(notBefore);
}
@Test
void shouldFetchAndUpdateBrokerStatisticsInGaugeRegistry() throws Exception {
SimpleGaugeRegistry gaugeRegistry = new SimpleGaugeRegistry();
ActiveMQMetricCollectorImpl testee = new ActiveMQMetricCollectorImpl(connectionFactory, new RecordingMetricFactory(), gaugeRegistry);
- ActiveMQBrokerStatistics brokerStatistics = new ActiveMQBrokerStatistics();
- brokerStatistics.registerMetrics(gaugeRegistry);
+ ActiveMQMetrics brokerStatistics = ActiveMQMetrics.forBroker(gaugeRegistry);
testee.fetchAndUpdate(brokerStatistics);
- Supplier<?> supplier = gaugeRegistry.getGauge("ActiveMQ.Statistics.Broker.storeLimit");
- assertThat(supplier.get()).isInstanceOf(Long.class);
- assertThat((Long) supplier.get()).isGreaterThan(0);
+ Number n = gaugeRegistry.getGauge("ActiveMQ.Statistics.Broker.storeLimit");
+ assertThat(n).isInstanceOf(Long.class);
+ assertThat((Long) n).isGreaterThan(0);
}
@Test
@@ -123,8 +123,17 @@ class ActiveMQMetricCollectorTest {
return this;
}
- public Gauge<?> getGauge(String name) {
- return gauges.get(name);
+ @Override
+ public <T> SettableGauge<T> settableGauge(String name) {
+ return t -> gauges.put(name, () -> t);
+ }
+
+ public Number getGauge(String name) {
+ Gauge<?> g = gauges.get(name);
+ if (g == null) {
+ return null;
+ }
+ return (Number) g.get();
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org