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 <noreply@anthropic.com>

Author: davsclaus

core/camel-management/src/main/java/org/apache/camel/management/mbean/LoadThroughput.java

@@ -19,18 +19,25 @@
 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 void update(long currentReading) {
             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 double getThroughput() {
     }
 
     public void reset() {
+        watch.stop();
         last = 0;
         thp = 0;
     }

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());
+    }
+
+}

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.

dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/EndpointsTab.java

@@ -444,17 +444,19 @@ private void renderEndpointFlow(
 
         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 @@ private void renderSingleEndpointChart(Frame frame, Rect area, String selectedUr
                 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 @@ public JsonObject getTableDataAsJson() {
         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;
+    }
+
 }

dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/MetricsCollector.java

@@ -19,6 +19,7 @@
 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 @@ void updateThroughputHistory(IntegrationInfo info) {
             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 @@ private void recordEndpointSample(
             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 @@ private void removeByPrefix(String prefix, Map<String, ?>... maps) {
             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";
+        }
+    }
 }