This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 62b024b3b1fd CAMEL-24032: Add rolling exchange throughput to base
performance counters
62b024b3b1fd is described below
commit 62b024b3b1fd34e4cab2b2c4b5e8f5d28a8a6e73
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jul 13 10:25:44 2026 +0200
CAMEL-24032: Add rolling exchange throughput to base performance counters
Consolidate LoadThroughput (EWMA-smoothed exchanges/second) from
ManagedRoute,
ManagedCamelContext, and ManagedRouteGroup into the base
ManagedPerformanceCounter.
All MBeans (including processor-level) now get throughput via TimerListener.
Adds throughput to TopDevConsole text and JSON output, and consolidates two
identical getStatsObject() methods into one using the base interface.
Closes #24645
Co-Authored-By: Claude <[email protected]>
---
.../apache/camel/impl/console/TopDevConsole.java | 57 ++++++++++------------
.../management/mbean/ManagedCamelContextMBean.java | 3 --
.../mbean/ManagedPerformanceCounterMBean.java | 3 ++
.../management/mbean/ManagedRouteGroupMBean.java | 3 --
.../api/management/mbean/ManagedRouteMBean.java | 3 --
.../management/mbean/ManagedCamelContext.java | 18 +------
.../mbean/ManagedPerformanceCounter.java | 22 ++++++++-
.../camel/management/mbean/ManagedRoute.java | 18 +------
.../camel/management/mbean/ManagedRouteGroup.java | 17 +------
9 files changed, 55 insertions(+), 89 deletions(-)
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/TopDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/TopDevConsole.java
index 88ed725daa3e..fab558b267ce 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/TopDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/TopDevConsole.java
@@ -92,6 +92,10 @@ public class TopDevConsole extends AbstractDevConsole {
String.format("%n Delta Time: %s",
TimeUtils.printDuration(mrb.getDeltaProcessingTime(), true)));
sb.append(
String.format("%n Total Time: %s",
TimeUtils.printDuration(mrb.getTotalProcessingTime(), true)));
+ String thp = mrb.getThroughput();
+ if (thp != null && !thp.isEmpty()) {
+ sb.append(String.format("%n Messages/Sec: %s",
thp));
+ }
sb.append("\n");
return null;
};
@@ -148,6 +152,10 @@ public class TopDevConsole extends AbstractDevConsole {
String.format("%n Delta Time: %s",
TimeUtils.printDuration(mpb.getDeltaProcessingTime(), true)));
sb.append(
String.format("%n Total Time: %s",
TimeUtils.printDuration(mpb.getTotalProcessingTime(), true)));
+ String thp = mpb.getThroughput();
+ if (thp != null && !thp.isEmpty()) {
+ sb.append(String.format("%n Messages/Sec: %s",
thp));
+ }
sb.append("\n");
return null;
};
@@ -246,40 +254,25 @@ public class TopDevConsole extends AbstractDevConsole {
return root;
}
- private static JsonObject getStatsObject(ManagedProcessorMBean mpb) {
+ private static JsonObject getStatsObject(ManagedPerformanceCounterMBean
mbean) {
JsonObject stats = new JsonObject();
- stats.put("exchangesTotal", mpb.getExchangesTotal());
- stats.put("exchangesFailed", mpb.getExchangesFailed());
- stats.put("exchangesInflight", mpb.getExchangesInflight());
- stats.put("meanProcessingTime", mpb.getMeanProcessingTime());
- stats.put("maxProcessingTime", mpb.getMaxProcessingTime());
- stats.put("minProcessingTime", mpb.getMinProcessingTime());
- stats.put("lastProcessingTime", mpb.getLastProcessingTime());
- stats.put("deltaProcessingTime", mpb.getDeltaProcessingTime());
- stats.put("totalProcessingTime", mpb.getTotalProcessingTime());
- if (mpb.getProcessingTimeP50() >= 0) {
- stats.put("p50ProcessingTime", mpb.getProcessingTimeP50());
- stats.put("p95ProcessingTime", mpb.getProcessingTimeP95());
- stats.put("p99ProcessingTime", mpb.getProcessingTimeP99());
+ stats.put("exchangesTotal", mbean.getExchangesTotal());
+ stats.put("exchangesFailed", mbean.getExchangesFailed());
+ stats.put("exchangesInflight", mbean.getExchangesInflight());
+ stats.put("meanProcessingTime", mbean.getMeanProcessingTime());
+ stats.put("maxProcessingTime", mbean.getMaxProcessingTime());
+ stats.put("minProcessingTime", mbean.getMinProcessingTime());
+ stats.put("lastProcessingTime", mbean.getLastProcessingTime());
+ stats.put("deltaProcessingTime", mbean.getDeltaProcessingTime());
+ stats.put("totalProcessingTime", mbean.getTotalProcessingTime());
+ String thp = mbean.getThroughput();
+ if (thp != null && !thp.isEmpty()) {
+ stats.put("exchangesThroughput", thp);
}
- return stats;
- }
-
- private static JsonObject getStatsObject(ManagedRouteMBean mrb) {
- JsonObject stats = new JsonObject();
- stats.put("exchangesTotal", mrb.getExchangesTotal());
- stats.put("exchangesFailed", mrb.getExchangesFailed());
- stats.put("exchangesInflight", mrb.getExchangesInflight());
- stats.put("meanProcessingTime", mrb.getMeanProcessingTime());
- stats.put("maxProcessingTime", mrb.getMaxProcessingTime());
- stats.put("minProcessingTime", mrb.getMinProcessingTime());
- stats.put("lastProcessingTime", mrb.getLastProcessingTime());
- stats.put("deltaProcessingTime", mrb.getDeltaProcessingTime());
- stats.put("totalProcessingTime", mrb.getTotalProcessingTime());
- if (mrb.getProcessingTimeP50() >= 0) {
- stats.put("p50ProcessingTime", mrb.getProcessingTimeP50());
- stats.put("p95ProcessingTime", mrb.getProcessingTimeP95());
- stats.put("p99ProcessingTime", mrb.getProcessingTimeP99());
+ if (mbean.getProcessingTimeP50() >= 0) {
+ stats.put("p50ProcessingTime", mbean.getProcessingTimeP50());
+ stats.put("p95ProcessingTime", mbean.getProcessingTimeP95());
+ stats.put("p99ProcessingTime", mbean.getProcessingTimeP99());
}
return stats;
}
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedCamelContextMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedCamelContextMBean.java
index d8936b269b57..9a27de8483f7 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedCamelContextMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedCamelContextMBean.java
@@ -135,9 +135,6 @@ public interface ManagedCamelContextMBean extends
ManagedPerformanceCounterMBean
@ManagedAttribute(description = "Average load (inflight messages, not cpu)
over the last fifteen minutes")
String getLoad15();
- @ManagedAttribute(description = "Throughput message/second")
- String getThroughput();
-
@ManagedAttribute(description = "Total number of exchanges processed from
remote endpoints only")
long getRemoteExchangesTotal();
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPerformanceCounterMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPerformanceCounterMBean.java
index e821f8be315b..d62da9e91f9d 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPerformanceCounterMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPerformanceCounterMBean.java
@@ -111,6 +111,9 @@ public interface ManagedPerformanceCounterMBean extends
ManagedCounterMBean {
@ManagedAttribute(description = "99th percentile of recent processing
times [milliseconds]. Requires Extended statistics level, returns -1
otherwise.")
long getProcessingTimeP99();
+ @ManagedAttribute(description = "Throughput (messages per second)")
+ String getThroughput();
+
@ManagedAttribute(description = "Statistics enabled")
boolean isStatisticsEnabled();
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteGroupMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteGroupMBean.java
index ff2b3f8486cf..37b6b8736435 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteGroupMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteGroupMBean.java
@@ -72,9 +72,6 @@ public interface ManagedRouteGroupMBean extends
ManagedPerformanceCounterMBean {
@ManagedAttribute(description = "Average load (inflight messages, not cpu)
over the last fifteen minutes")
String getLoad15();
- @ManagedAttribute(description = "Throughput message/second")
- String getThroughput();
-
@ManagedAttribute(description = "Last error from the routes in this group")
RouteError getLastError();
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteMBean.java
index fa3f1abb3397..90473b64a932 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedRouteMBean.java
@@ -103,9 +103,6 @@ public interface ManagedRouteMBean extends
ManagedPerformanceCounterMBean {
@ManagedAttribute(description = "Average load (inflight messages, not cpu)
over the last fifteen minutes")
String getLoad15();
- @ManagedAttribute(description = "Throughput message/second")
- String getThroughput();
-
@ManagedOperation(description = "Start route")
void start() throws Exception;
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
index 751cbabb3d03..3bbcfe8b7805 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
@@ -37,7 +37,6 @@ import org.apache.camel.ManagementStatisticsLevel;
import org.apache.camel.Producer;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.Route;
-import org.apache.camel.TimerListener;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.api.management.mbean.ManagedCamelContextMBean;
import org.apache.camel.api.management.mbean.ManagedProcessorMBean;
@@ -58,11 +57,10 @@ import org.apache.camel.util.json.JsonArray;
import org.apache.camel.util.json.JsonObject;
@ManagedResource(description = "Managed CamelContext")
-public class ManagedCamelContext extends ManagedPerformanceCounter implements
TimerListener, ManagedCamelContextMBean {
+public class ManagedCamelContext extends ManagedPerformanceCounter implements
ManagedCamelContextMBean {
private final CamelContext context;
private final LoadTriplet load = new LoadTriplet();
- private final LoadThroughput thp = new LoadThroughput();
private final String jmxDomain;
private final boolean includeRouteTemplates;
private final boolean includeKamelets;
@@ -383,17 +381,6 @@ public class ManagedCamelContext extends
ManagedPerformanceCounter implements Ti
}
}
- @Override
- public String getThroughput() {
- double d = thp.getThroughput();
- if (Double.isNaN(d)) {
- // empty string if load statistics is disabled
- return "";
- } else {
- return String.format("%.2f", d);
- }
- }
-
@Override
public long getRemoteExchangesTotal() {
return remoteExchangesTotal.getValue();
@@ -447,8 +434,8 @@ public class ManagedCamelContext extends
ManagedPerformanceCounter implements Ti
@Override
public void onTimer() {
+ super.onTimer();
load.update(getInflightExchanges());
- thp.update(getExchangesTotal());
}
@Override
@@ -976,7 +963,6 @@ public class ManagedCamelContext extends
ManagedPerformanceCounter implements Ti
public void reset(boolean includeRoutes) throws Exception {
reset();
load.reset();
- thp.reset();
// and now reset all routes for this route
if (includeRoutes) {
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
index e51e60565faa..6b7187eb27ef 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
@@ -22,6 +22,7 @@ import java.util.Date;
import java.util.Map;
import org.apache.camel.Exchange;
+import org.apache.camel.TimerListener;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.api.management.mbean.ManagedPerformanceCounterMBean;
import org.apache.camel.management.PerformanceCounter;
@@ -31,7 +32,7 @@ import org.apache.camel.util.json.JsonObject;
@ManagedResource(description = "Managed PerformanceCounter")
public abstract class ManagedPerformanceCounter extends ManagedCounter
- implements PerformanceCounter, ManagedPerformanceCounterMBean {
+ implements TimerListener, PerformanceCounter,
ManagedPerformanceCounterMBean {
public static final String TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
@@ -59,6 +60,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
private Statistic lastExchangeFailureHandledTimestamp;
private Statistic lastExchangeFailureTimestamp;
private String lastExchangeFailureExchangeId;
+ private final LoadThroughput thp = new LoadThroughput();
private boolean statisticsEnabled = true;
// sliding window ring buffer for percentile computation (Extended
statistics only)
@@ -123,6 +125,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
lastExchangeFailureHandledTimestamp.reset();
lastExchangeFailureTimestamp.reset();
lastExchangeFailureExchangeId = null;
+ thp.reset();
if (percentileWindow != null) {
percentileIndex = 0;
percentileCount = 0;
@@ -286,6 +289,21 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
this.statisticsEnabled = statisticsEnabled;
}
+ @Override
+ public String getThroughput() {
+ double d = thp.getThroughput();
+ if (Double.isNaN(d)) {
+ return "";
+ } else {
+ return String.format("%.2f", d);
+ }
+ }
+
+ @Override
+ public void onTimer() {
+ thp.update(getExchangesTotal());
+ }
+
@Override
public void processExchange(Exchange exchange, String type) {
exchangesInflight.increment();
@@ -395,6 +413,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
sb.append(String.format(" lastProcessingTime=\"%s\"",
lastProcessingTime.getValue()));
sb.append(String.format(" deltaProcessingTime=\"%s\"",
deltaProcessingTime.getValue()));
sb.append(String.format(" meanProcessingTime=\"%s\"",
meanProcessingTime.getValue()));
+ sb.append(String.format(" exchangesThroughput=\"%s\"",
getThroughput()));
if (percentileWindow != null) {
sb.append(String.format(" p50ProcessingTime=\"%s\"",
getProcessingTimeP50()));
sb.append(String.format(" p95ProcessingTime=\"%s\"",
getProcessingTimeP95()));
@@ -447,6 +466,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
jo.put("lastProcessingTime", lastProcessingTime.getValue());
jo.put("deltaProcessingTime", deltaProcessingTime.getValue());
jo.put("meanProcessingTime", meanProcessingTime.getValue());
+ jo.put("exchangesThroughput", getThroughput());
if (percentileWindow != null) {
jo.put("p50ProcessingTime", getProcessingTimeP50());
jo.put("p95ProcessingTime", getProcessingTimeP95());
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
index 86cd24c50af2..e64d74cd4882 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
@@ -48,7 +48,6 @@ import org.apache.camel.ManagementStatisticsLevel;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.ServiceStatus;
-import org.apache.camel.TimerListener;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.api.management.mbean.CamelOpenMBeanTypes;
import org.apache.camel.api.management.mbean.ManagedProcessorMBean;
@@ -71,7 +70,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ManagedResource(description = "Managed Route")
-public class ManagedRoute extends ManagedPerformanceCounter implements
TimerListener, ManagedRouteMBean {
+public class ManagedRoute extends ManagedPerformanceCounter implements
ManagedRouteMBean {
public static final String VALUE_UNKNOWN = "Unknown";
@@ -85,7 +84,6 @@ public class ManagedRoute extends ManagedPerformanceCounter
implements TimerList
protected final String sourceLocationShort;
protected final CamelContext context;
private final LoadTriplet load = new LoadTriplet();
- private final LoadThroughput thp = new LoadThroughput();
private final String jmxDomain;
public ManagedRoute(CamelContext context, Route route) {
@@ -315,21 +313,10 @@ public class ManagedRoute extends
ManagedPerformanceCounter implements TimerList
}
}
- @Override
- public String getThroughput() {
- double d = thp.getThroughput();
- if (Double.isNaN(d)) {
- // empty string if load statistics is disabled
- return "";
- } else {
- return String.format("%.2f", d);
- }
- }
-
@Override
public void onTimer() {
+ super.onTimer();
load.update(getInflightExchanges());
- thp.update(getExchangesTotal());
}
@Override
@@ -842,7 +829,6 @@ public class ManagedRoute extends ManagedPerformanceCounter
implements TimerList
public void reset(boolean includeProcessors) throws Exception {
reset();
load.reset();
- thp.reset();
// and now reset all processors for this route
if (includeProcessors) {
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java
index efcf5aadffc9..fc42869151c0 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java
@@ -23,7 +23,6 @@ import org.apache.camel.CamelContext;
import org.apache.camel.ManagementStatisticsLevel;
import org.apache.camel.Route;
import org.apache.camel.ServiceStatus;
-import org.apache.camel.TimerListener;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.api.management.mbean.ManagedRouteGroupMBean;
import org.apache.camel.api.management.mbean.RouteError;
@@ -31,14 +30,13 @@ import org.apache.camel.spi.ManagementStrategy;
import org.apache.camel.util.TimeUtils;
@ManagedResource(description = "Managed Route Group")
-public class ManagedRouteGroup extends ManagedPerformanceCounter implements
TimerListener, ManagedRouteGroupMBean {
+public class ManagedRouteGroup extends ManagedPerformanceCounter implements
ManagedRouteGroupMBean {
public static final String VALUE_UNKNOWN = "Unknown";
protected final String group;
protected final CamelContext context;
private final LoadTriplet load = new LoadTriplet();
- private final LoadThroughput thp = new LoadThroughput();
private final String jmxDomain;
public ManagedRouteGroup(CamelContext context, String group) {
@@ -155,17 +153,6 @@ public class ManagedRouteGroup extends
ManagedPerformanceCounter implements Time
}
}
- @Override
- public String getThroughput() {
- double d = thp.getThroughput();
- if (Double.isNaN(d)) {
- // empty string if load statistics is disabled
- return "";
- } else {
- return String.format("%.2f", d);
- }
- }
-
@Override
public RouteError getLastError() {
org.apache.camel.spi.RouteError last = null;
@@ -222,8 +209,8 @@ public class ManagedRouteGroup extends
ManagedPerformanceCounter implements Time
@Override
public void onTimer() {
+ super.onTimer();
load.update(getInflightExchanges());
- thp.update(getExchangesTotal());
}
private Integer getInflightExchanges() {
