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 5c4680344812 CAMEL-23865: Use EWMA smoothing for throughput and fix
TUI sparkline scaling
5c4680344812 is described below
commit 5c468034481263351bb5607ccc8d331c19f2cfde
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 1 18:19:33 2026 +0200
CAMEL-23865: Use EWMA smoothing for throughput and fix TUI sparkline scaling
Switch the throughput metric in LoadThroughput from raw instantaneous rate
to
EWMA smoothing with a 1-minute decay window. Also fixes TUI sparkline charts
to properly display sub-1.0 msg/s rates by scaling throughput values by
100x.
Co-Authored-By: Claude <[email protected]>
---
.../camel/management/mbean/LoadThroughput.java | 26 +++---
.../camel/management/LoadThroughputTest.java | 93 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 10 +++
.../dsl/jbang/core/commands/tui/EndpointsTab.java | 29 +++++--
.../jbang/core/commands/tui/MetricsCollector.java | 58 ++++++++++++--
.../dsl/jbang/core/commands/tui/OverviewTab.java | 32 +++++---
6 files changed, 214 insertions(+), 34 deletions(-)
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/LoadThroughput.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/LoadThroughput.java
index dd05d4595d87..89c60f806e28 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/LoadThroughput.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/LoadThroughput.java
@@ -19,18 +19,25 @@ package org.apache.camel.management.mbean;
import org.apache.camel.util.StopWatch;
/**
- * Holds the load throughput messages/second
+ * Holds the throughput (messages/second) using EWMA (exponentially weighted
moving average) smoothing, modeled after
+ * Unix load averages (same approach as {@link LoadTriplet}).
+ *
+ * The instantaneous rate from each 1-second sampling interval is smoothed
with a 1-minute decay window so that the
+ * reported value converges to the true average rate instead of oscillating
between 0 and spike values.
*/
public final class LoadThroughput {
+ // EWMA exponent for a 1-minute decay window, sampled every 1 second
+ private static final double EXP_1 = Math.exp(-1.0 / 60.0);
+
private final StopWatch watch = new StopWatch(false);
private long last;
private double thp;
/**
- * Update the load statistics
+ * Update the throughput statistics
*
- * @param currentReading the current reading
+ * @param currentReading the current cumulative exchange count
*/
public void update(long currentReading) {
if (!watch.isStarted()) {
@@ -40,14 +47,10 @@ public final class LoadThroughput {
long time = watch.takenAndRestart();
if (time > 0) {
long delta = currentReading - last;
- if (delta > 0) {
- // need to calculate with fractions
- thp = (1000d / time) * delta;
- } else {
- thp = 0;
- }
- } else {
- thp = 0;
+ // instantaneous rate in exchanges/second for this interval
+ double instantRate = Math.max(0, (1000d / time) * delta);
+ // apply EWMA smoothing
+ thp = instantRate + EXP_1 * (thp - instantRate);
}
}
last = currentReading;
@@ -58,6 +61,7 @@ public final class LoadThroughput {
}
public void reset() {
+ watch.stop();
last = 0;
thp = 0;
}
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/LoadThroughputTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/LoadThroughputTest.java
new file mode 100644
index 000000000000..0234748823fe
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/LoadThroughputTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.camel.management;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.management.mbean.LoadThroughput;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LoadThroughputTest {
+
+ @Test
+ public void testInitialValueIsZero() {
+ LoadThroughput t = new LoadThroughput();
+ assertEquals(0.0, t.getThroughput());
+ }
+
+ @Test
+ public void testConvergesToSteadyRate() {
+ LoadThroughput t = new LoadThroughput();
+ AtomicLong total = new AtomicLong(0);
+ t.update(total.get());
+
+ await().pollInterval(10, TimeUnit.MILLISECONDS)
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ total.incrementAndGet();
+ t.update(total.get());
+ assertTrue(t.getThroughput() > 5.0,
+ "Throughput should converge toward steady rate: "
+ t.getThroughput());
+ });
+ }
+
+ @Test
+ public void testSmoothing() {
+ LoadThroughput t = new LoadThroughput();
+ AtomicLong total = new AtomicLong(0);
+ AtomicInteger count = new AtomicInteger(0);
+ t.update(total.get());
+
+ await().pollInterval(10, TimeUnit.MILLISECONDS)
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ int i = count.incrementAndGet();
+ if (i % 5 == 0) {
+ total.incrementAndGet();
+ }
+ t.update(total.get());
+ double thp = t.getThroughput();
+ assertTrue(thp > 1.0, "Smoothed throughput should be well
above zero: " + thp);
+ assertTrue(thp < 80.0, "Smoothed throughput should be
below the instantaneous spike: " + thp);
+ });
+ }
+
+ @Test
+ public void testReset() {
+ LoadThroughput t = new LoadThroughput();
+ AtomicLong total = new AtomicLong(0);
+ t.update(total.get());
+
+ await().pollInterval(10, TimeUnit.MILLISECONDS)
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ total.addAndGet(10);
+ t.update(total.get());
+ assertTrue(t.getThroughput() > 0);
+ });
+
+ t.reset();
+ assertEquals(0.0, t.getThroughput());
+ }
+
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 3d5efab27be2..4f27d41e9d39 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -30,6 +30,16 @@ String chat(AiAgentBody<?> aiAgentBody, ToolProvider
toolProvider);
Result<String> chat(AiAgentBody<?> aiAgentBody, ToolProvider toolProvider);
----
+=== camel-management - Throughput MBean attribute uses EWMA smoothing
+
+The `Throughput` attribute on the `ManagedPerformanceCounter` JMX MBean now
reports an EWMA
+(exponentially weighted moving average) value with a 1-minute decay window,
instead of the
+previous raw instantaneous rate. This produces a smoother, more stable reading
that converges
+to the true average throughput rather than oscillating between zero and spike
values.
+
+If you have tooling that consumes the JMX throughput value and expects the old
instantaneous
+behavior, be aware that the reported value will now ramp up and decay
gradually.
+
=== camel-fory with JDK 25+ - Breaking change
Due to new requirements from Apache Fory, when using Apache Fory Dataformat,
the JVM parameter `--add-opens java.base/java.lang.invoke=ALL-UNNAMED` must be
provided.
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/EndpointsTab.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/EndpointsTab.java
index e50c0b0acb79..c815bdeb0bed 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/EndpointsTab.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/EndpointsTab.java
@@ -444,17 +444,19 @@ class EndpointsTab implements MonitorTab {
Line chartTitle = Line.from(
Span.styled("▬",
Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN))),
- Span.raw(String.format(" in:%-4d ", curIn)),
+ Span.raw(String.format(" in:%-4s ",
MetricsCollector.formatThroughput(curIn))),
Span.styled("▬", Style.EMPTY.fg(Color.CYAN)),
- Span.raw(String.format(" out:%-4d msg/s", curOut)));
+ Span.raw(String.format(" out:%-4s msg/s",
MetricsCollector.formatThroughput(curOut))));
Rect rightArea = hSplit.get(1);
+ long maxEp = MetricsCollector.niceMax(Math.max(maxOf(inArr),
maxOf(outArr)));
frame.renderWidget(DualSparkline.builder()
.topData(inArr)
.bottomData(outArr)
+ .max(maxEp)
.topStyle(Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN)))
.bottomStyle(Style.EMPTY.fg(Color.CYAN))
- .showYAxis(true)
+ .showYAxis(false)
.xLabels("-" + renderPoints + "s", "-" + (renderPoints * 3 /
4) + "s",
"-" + (renderPoints / 2) + "s", "-" + (renderPoints /
4) + "s", "now")
.block(Block.builder().borderType(BorderType.ROUNDED).borders(Borders.ALL)
@@ -555,16 +557,20 @@ class EndpointsTab implements MonitorTab {
Span.styled(uriLabel, Style.EMPTY.fg(Color.YELLOW)),
Span.raw("] "),
Span.styled("▬",
Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN))),
- Span.raw(String.format(" in:%-4d ", curIn)),
+ Span.raw(String.format(" in:%-4s ",
MetricsCollector.formatThroughput(curIn))),
Span.styled("▬", Style.EMPTY.fg(Color.CYAN)),
- Span.raw(String.format(" out:%-4d msg/s", curOut)));
+ Span.raw(String.format(" out:%-4s msg/s",
MetricsCollector.formatThroughput(curOut))));
+ // TODO: use
.showYAxis(true).yAxisFormatter(MetricsCollector::formatThroughput) when
tamboui 0.5.0 is released
+ // see https://github.com/tamboui/tamboui/pull/396
+ long maxEpSingle = MetricsCollector.niceMax(Math.max(maxOf(inArr),
maxOf(outArr)));
frame.renderWidget(DualSparkline.builder()
.topData(inArr)
.bottomData(outArr)
+ .max(maxEpSingle)
.topStyle(Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN)))
.bottomStyle(Style.EMPTY.fg(Color.CYAN))
- .showYAxis(true)
+ .showYAxis(false)
.xLabels("-" + renderPoints + "s", "-" + (renderPoints * 3 /
4) + "s",
"-" + (renderPoints / 2) + "s", "-" + (renderPoints /
4) + "s", "now")
.block(Block.builder().borderType(BorderType.ROUNDED).borders(Borders.ALL)
@@ -770,4 +776,15 @@ class EndpointsTab implements MonitorTab {
result.put("selectedIndex", sel != null ? sel : -1);
return result;
}
+
+ private static long maxOf(long[] arr) {
+ long max = 0;
+ for (long v : arr) {
+ if (v > max) {
+ max = v;
+ }
+ }
+ return max;
+ }
+
}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/MetricsCollector.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/MetricsCollector.java
index 9670b766d22f..df9e43db3788 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/MetricsCollector.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/MetricsCollector.java
@@ -19,6 +19,7 @@ package org.apache.camel.dsl.jbang.core.commands.tui;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.LinkedList;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@@ -33,6 +34,8 @@ class MetricsCollector {
static final int MAX_ENDPOINT_CHART_POINTS = 300;
static final int MAX_HEAP_HISTORY_POINTS = 120;
static final long HEAP_SAMPLE_INTERVAL_MS = 5000;
+ // Throughput values are stored scaled by this factor so sub-1.0 msg/s
rates are preserved as longs
+ static final long THROUGHPUT_SCALE = 100;
// Throughput history per PID (one point per second)
private final Map<String, LinkedList<Long>> throughputHistory = new
ConcurrentHashMap<>();
@@ -164,15 +167,28 @@ class MetricsCollector {
samples.remove(0);
}
+ // Use the EWMA throughput from the status JSON (already smoothed in
camel-core)
+ // and store scaled by 100 so sub-1.0 rates (e.g. 0.20 msg/s) are
preserved as integers
+ long tp = 0;
+ if (info.throughput != null) {
+ try {
+ tp = Math.round(Double.parseDouble(info.throughput) *
THROUGHPUT_SCALE);
+ } catch (NumberFormatException e) {
+ // ignore
+ }
+ }
+
+ // Failed throughput still computed from delta (no EWMA source for
failed-only)
+ long fp = 0;
if (samples.size() >= 2) {
long[] oldest = samples.get(0);
long[] newest = samples.get(samples.size() - 1);
- long deltaTotal = newest[1] - oldest[1];
long deltaFailed = newest[2] - oldest[2];
long deltaTimeMs = newest[0] - oldest[0];
- long tp = deltaTimeMs > 0 ? (deltaTotal * 1000) / deltaTimeMs : 0;
- long fp = deltaTimeMs > 0 ? (deltaFailed * 1000) / deltaTimeMs : 0;
+ fp = deltaTimeMs > 0 ? (deltaFailed * 1000 * THROUGHPUT_SCALE) /
deltaTimeMs : 0;
+ }
+ {
Long lastTime = previousExchangesTime.get(pid);
if (lastTime == null || now - lastTime >= 1000) {
previousExchangesTime.put(pid, now);
@@ -262,8 +278,8 @@ class MetricsCollector {
long[] oldest = samples.get(0);
long[] newest = samples.get(samples.size() - 1);
long deltaMs = newest[0] - oldest[0];
- long inRate = deltaMs > 0 ? (newest[1] - oldest[1]) * 1000 /
deltaMs : 0;
- long outRate = deltaMs > 0 ? (newest[2] - oldest[2]) * 1000 /
deltaMs : 0;
+ long inRate = deltaMs > 0 ? (newest[1] - oldest[1]) * 1000 *
THROUGHPUT_SCALE / deltaMs : 0;
+ long outRate = deltaMs > 0 ? (newest[2] - oldest[2]) * 1000 *
THROUGHPUT_SCALE / deltaMs : 0;
Long lastTime = prevTimeMap.get(pid);
if (lastTime == null || now - lastTime >= 1000) {
prevTimeMap.put(pid, now);
@@ -403,4 +419,36 @@ class MetricsCollector {
map.keySet().removeIf(k -> k.startsWith(prefix));
}
}
+
+ // --- Shared throughput formatting utilities ---
+
+ static long niceMax(long rawMax) {
+ if (rawMax <= 0) {
+ return THROUGHPUT_SCALE;
+ }
+ int[] steps = { 1, 2, 5 };
+ long multiplier = THROUGHPUT_SCALE;
+ while (true) {
+ for (int s : steps) {
+ long candidate = s * multiplier;
+ if (candidate >= rawMax) {
+ return candidate;
+ }
+ }
+ multiplier *= 10;
+ }
+ }
+
+ static String formatThroughput(long scaledValue) {
+ double v = scaledValue / (double) THROUGHPUT_SCALE;
+ if (v >= 10) {
+ return String.valueOf(Math.round(v));
+ } else if (v >= 1) {
+ return String.format(Locale.US, "%.1f", v);
+ } else if (scaledValue > 0) {
+ return String.format(Locale.US, "%.2f", v);
+ } else {
+ return "0";
+ }
+ }
}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OverviewTab.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OverviewTab.java
index ca705192d82f..39979db1ccf5 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OverviewTab.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OverviewTab.java
@@ -227,7 +227,7 @@ class OverviewTab implements MonitorTab {
List<Constraint> constraints = new ArrayList<>();
constraints.add(Constraint.fill());
if (hasSparkline) {
- constraints.add(Constraint.length(14));
+ constraints.add(Constraint.length(17));
} else if (showInfoPanel) {
int panelH = countInfraLines(infraSel) + 2;
constraints.add(Constraint.length(Math.min(panelH, area.height() /
2)));
@@ -416,7 +416,7 @@ class OverviewTab implements MonitorTab {
.split(chartInner);
List<Rect> hChunks = Layout.horizontal()
- .constraints(Constraint.length(4), Constraint.fill())
+ .constraints(Constraint.length(5), Constraint.fill())
.split(vChunks.get(0));
Rect barChartArea = hChunks.get(1);
@@ -450,10 +450,16 @@ class OverviewTab implements MonitorTab {
for (long v : mergedTotal) {
maxTp = Math.max(maxTp, v);
}
+ maxTp = MetricsCollector.niceMax(maxTp);
long curTp = mergedTotal[renderPoints - 1];
long curFailed = mergedFailed[renderPoints - 1];
long curOk = Math.max(0, curTp - curFailed);
+ // Format throughput values unscaled for display
+ String curTpFmt = MetricsCollector.formatThroughput(curTp);
+ String curOkFmt = MetricsCollector.formatThroughput(curOk);
+ String curFailFmt = MetricsCollector.formatThroughput(curFailed);
+
Line titleLine;
if (chartMode == CHART_SINGLE && ctx.selectedPid != null) {
IntegrationInfo chartSel = ctx.findSelectedIntegration();
@@ -462,18 +468,18 @@ class OverviewTab implements MonitorTab {
titleLine = Line.from(
Span.raw(" ["),
Span.styled(chartName, Style.EMPTY.fg(Color.YELLOW)),
- Span.raw(String.format("] Throughput: %d msg/s ",
curTp)),
+ Span.raw(String.format("] Throughput: %s msg/s ",
curTpFmt)),
Span.styled("■",
Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN))),
- Span.raw(String.format(" ok:%d ", curOk)),
+ Span.raw(String.format(" ok:%s ", curOkFmt)),
Span.styled("■", Style.EMPTY.fg(Color.RED)),
- Span.raw(String.format(" fail:%d ", curFailed)));
+ Span.raw(String.format(" fail:%s ", curFailFmt)));
} else {
titleLine = Line.from(
- Span.raw(String.format(" [All] Throughput: %d msg/s
", curTp)),
+ Span.raw(String.format(" [All] Throughput: %s msg/s
", curTpFmt)),
Span.styled("■",
Style.EMPTY.fg(Color.ansi(AnsiColor.BRIGHT_GREEN))),
- Span.raw(String.format(" ok:%d ", curOk)),
+ Span.raw(String.format(" ok:%s ", curOkFmt)),
Span.styled("■", Style.EMPTY.fg(Color.RED)),
- Span.raw(String.format(" fail:%d ", curFailed)));
+ Span.raw(String.format(" fail:%s ", curFailFmt)));
}
Block chartBlock =
Block.builder().borderType(BorderType.ROUNDED).borders(Borders.ALL)
@@ -492,7 +498,7 @@ class OverviewTab implements MonitorTab {
BarChart barChart = BarChart.builder()
.data(groups)
- .max(maxTp > 0 ? maxTp + 2 : 2)
+ .max(maxTp)
.barWidth(1)
.barGap(0)
.groupGap(0)
@@ -505,11 +511,13 @@ class OverviewTab implements MonitorTab {
Style dimStyle = Style.EMPTY.dim();
for (int row = 0; row < barRows; row++) {
if (row == 0) {
- yLines.add(Line.from(Span.styled(String.format("%3d",
maxTp), dimStyle)));
+ yLines.add(
+ Line.from(Span.styled(String.format("%4s",
MetricsCollector.formatThroughput(maxTp)), dimStyle)));
} else if (barRows > 4 && row == barRows / 2) {
- yLines.add(Line.from(Span.styled(String.format("%3d",
maxTp / 2), dimStyle)));
+ yLines.add(Line
+ .from(Span.styled(String.format("%4s",
MetricsCollector.formatThroughput(maxTp / 2)), dimStyle)));
} else if (row == barRows - 1) {
- yLines.add(Line.from(Span.styled(" 0", dimStyle)));
+ yLines.add(Line.from(Span.styled(" 0", dimStyle)));
} else {
yLines.add(Line.from(""));
}
