This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24032 in repository https://gitbox.apache.org/repos/asf/camel.git
commit f150e70e45cf1676f6c46430f46736c62014dcc1 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Jul 13 09:34:07 2026 +0200 CAMEL-24032: Add rolling exchange throughput to base performance counters Co-Authored-By: Claude <[email protected]> Signed-off-by: Claus Ibsen <[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() {
