This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 8785360b01 feat(loadbalancer): expose health-check round observability
metrics (#7647) (#7674)
8785360b01 is described below
commit 8785360b01760fa4de55d4615998f35de239d376
Author: Daming <[email protected]>
AuthorDate: Tue Oct 20 12:39:15 2026 +0800
feat(loadbalancer): expose health-check round observability metrics (#7647)
(#7674)
* feat(loadbalancer): expose health-check round observability metrics
(#7647)
UpstreamCheckTask schedules fixed-delay rounds, submits every registered
upstream to a separate executor and waits for the round inside a
synchronized section. Large batches or slow endpoints can increase round
duration and delay subsequent checks, but today there is no way to tell
from the outside whether a round is falling behind the configured
checkInterval, whether the executor or the endpoints are the bottleneck,
or how stale the published health state actually is.
Issue #7647 asks for these signals to be recorded before any scheduling
change is evaluated. This commit adds the observability layer only; the
scheduling, locking, healthy/unhealthy thresholds, update/removal races
and failure-recovery behavior are unchanged.
New per-round signals (all readable via public getters):
- lastRoundDurationMillis: monotonic System.nanoTime elapsed across the
whole synchronized round (submit + wait), so it reflects real time to
converge rather than publisher-construction time.
- lastRoundSubmittedChecks: how many probes were dispatched this round.
- lastRoundPendingChecks: how many of those were still pending when the
round entered waitFinish; a persistently high value points at the
executor, not the endpoints.
- lastRoundMaxEndpointLatencyMillis: worst single-probe latency observed
this round (only probes with health-check enabled contribute).
- lastRoundOldestHealthStateMillis: max age across healthy and unhealthy
buckets, computed from Upstream.lastHealthTimestamp /
lastUnhealthyTimestamp; grows when rounds fall behind checkInterval.
- lastRoundCompletedAtMillis: wall-clock finish time.
Cumulative counters (totalRounds, totalSubmittedChecks,
totalEndpointLatencyMillis) let operators compute long-run averages
without a metrics backend.
A DEBUG-level per-round summary line is emitted so the log volume stays
aligned with #7643; no INFO line is added per round.
Tests (UpstreamCheckTaskTest):
- roundMetricsAreRecordedAfterRunCompletes: after one synchronous run()
with four registered upstreams, all last-round fields are populated
and submitted covers every upstream.
- emptyPopulationStillRecordsRoundCompletion: totalRounds advances even
with zero upstreams, so operators can distinguish 'not running' from
'running with nothing to check'; endpoint latency stays zero.
- oldestHealthStateAgeReflectsStaleTimestamp: an upstream whose
lastHealthTimestamp is 5s old surfaces as an oldest-age of at least
4s, so the metric does not collapse to zero.
- perRoundCountersAreResetBetweenRounds: a second run() resets the
per-round scratch counters instead of leaking values from the previous
round.
Ref: #7650
* refactor(loadbalancer): scope health-check round counters to locals
Replace the currentRoundSubmittedChecks and
currentRoundMaxEndpointLatencyMillis
fields with per-round locals in healthCheck(). The submitted count is now
returned
by doHealthCheck()/check(Map), while the max endpoint latency stays an
AtomicLong
passed as a parameter so the async probe tasks can still update it
concurrently
via the captured final reference.
* refactor(loadbalancer): simplify doHealthCheck submitted count
Collapse the local accumulator in doHealthCheck() into a single return
expression summing the healthy and unhealthy check counts.
* fix(loadbalancer): skip never-measured upstreams in health-state age
(#7647)
computeOldestHealthStateMillis() treated an unset (0)
lastHealth/lastUnhealthy
timestamp as an epoch-sized age, so a single never-probed upstream (newly
registered, or health-check-disabled while already healthy) dominated the
metric
and defeated its purpose. Guard both buckets with timestamp > 0 so
unmeasured
entries are skipped and an all-unmeasured population reports 0.
Also make the stale-age test use a blank URL instead of a fixed port
(65100) so an
unrelated listener cannot flip the probe to pass, add a regression test for
the
disabled+healthy zero-timestamp case, and document the skipped-round
semantics of
getLastRoundDurationMillis()/getLastRoundSubmittedChecks().
---------
Co-authored-by: Luke Haochao Zhuang <[email protected]>
---
.../loadbalancer/cache/UpstreamCheckTask.java | 213 ++++++++++++++++++++-
.../loadbalancer/cache/UpstreamCheckTaskTest.java | 152 +++++++++++++++
2 files changed, 357 insertions(+), 8 deletions(-)
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java
index 114aff74c0..a37976667e 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java
@@ -39,6 +39,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
@@ -72,6 +73,24 @@ public final class UpstreamCheckTask implements Runnable {
private int healthyThreshold = 1;
private int unhealthyThreshold = 1;
+
+ private volatile long lastRoundDurationMillis;
+
+ private volatile int lastRoundSubmittedChecks;
+
+ private volatile int lastRoundPendingChecks;
+
+ private volatile long lastRoundMaxEndpointLatencyMillis;
+
+ private volatile long lastRoundOldestHealthStateMillis;
+
+ private volatile long lastRoundCompletedAtMillis;
+
+ private final AtomicLong totalRounds = new AtomicLong();
+
+ private final AtomicLong totalSubmittedChecks = new AtomicLong();
+
+ private final AtomicLong totalEndpointLatencyMillis = new AtomicLong();
/**
* Instantiates a new Upstream check task.
@@ -156,6 +175,12 @@ public final class UpstreamCheckTask implements Runnable {
}
private void healthCheck() {
+ final long startNanos = System.nanoTime();
+ // Per-round scratch counters. maxEndpointLatencyMillis is mutated by
the async probe
+ // tasks running on the executor, so it stays an AtomicLong captured
by the lambdas.
+ final AtomicLong maxEndpointLatencyMillis = new AtomicLong();
+ int submitted = 0;
+ int pendingAtWait = 0;
try {
/*
* If there is no synchronized. when check is done and all
upstream check result is in the futures list.
@@ -163,7 +188,8 @@ public final class UpstreamCheckTask implements Runnable {
*/
synchronized (lock) {
if (tryStartHealthCheck()) {
- doHealthCheck();
+ submitted = doHealthCheck(maxEndpointLatencyMillis);
+ pendingAtWait = countPendingFutures();
waitFinish();
}
}
@@ -171,26 +197,94 @@ public final class UpstreamCheckTask implements Runnable {
LOG.error("[Health Check] Meet problem: ", e);
} finally {
finishHealthCheck();
+ recordRoundMetrics(startNanos, submitted, pendingAtWait,
maxEndpointLatencyMillis.get());
}
}
- private void doHealthCheck() {
- check(healthyUpstream);
- check(unhealthyUpstream);
+ private int countPendingFutures() {
+ int pending = 0;
+ for (CompletableFuture<UpstreamWithSelectorId> future : futures) {
+ if (!future.isDone()) {
+ pending++;
+ }
+ }
+ return pending;
+ }
+
+ private void recordRoundMetrics(final long startNanos, final int
submitted, final int pendingAtWait,
+ final long maxEndpointLatencyMillis) {
+ final long durationMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ final long oldestHealthStateMillis = computeOldestHealthStateMillis();
+ final long completedAtMillis = System.currentTimeMillis();
+ this.lastRoundDurationMillis = durationMillis;
+ this.lastRoundSubmittedChecks = submitted;
+ this.lastRoundPendingChecks = pendingAtWait;
+ this.lastRoundMaxEndpointLatencyMillis = maxEndpointLatencyMillis;
+ this.lastRoundOldestHealthStateMillis = oldestHealthStateMillis;
+ this.lastRoundCompletedAtMillis = completedAtMillis;
+ totalRounds.incrementAndGet();
+ totalSubmittedChecks.addAndGet(submitted);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("[Health Check] round summary: durationMs={},
submitted={}, pendingAtWait={}, "
+ + "maxEndpointLatencyMs={},
oldestHealthStateMs={}, healthySelectors={}, unhealthySelectors={}",
+ durationMillis, submitted, pendingAtWait,
maxEndpointLatencyMillis, oldestHealthStateMillis,
+ healthyUpstream.size(), unhealthyUpstream.size());
+ }
}
- private void check(final Map<String, List<Upstream>> map) {
+ private long computeOldestHealthStateMillis() {
+ final long now = System.currentTimeMillis();
+ long oldest = 0L;
+ for (List<Upstream> bucket : healthyUpstream.values()) {
+ for (Upstream upstream : bucket) {
+ // Skip never-measured upstreams: lastHealthTimestamp defaults
to 0 for entries that
+ // were registered but not probed yet, or whose health check
is disabled while already
+ // healthy. Without this guard now - 0 would report an
epoch-sized (since 1970) age and
+ // permanently dominate the metric.
+ final long timestamp = upstream.getLastHealthTimestamp();
+ if (timestamp > 0) {
+ final long age = now - timestamp;
+ if (age > oldest) {
+ oldest = age;
+ }
+ }
+ }
+ }
+ for (List<Upstream> bucket : unhealthyUpstream.values()) {
+ for (Upstream upstream : bucket) {
+ final long timestamp = upstream.getLastUnhealthyTimestamp();
+ if (timestamp > 0) {
+ final long age = now - timestamp;
+ if (age > oldest) {
+ oldest = age;
+ }
+ }
+ }
+ }
+ return oldest;
+ }
+
+ private int doHealthCheck(final AtomicLong maxEndpointLatencyMillis) {
+ return check(healthyUpstream, maxEndpointLatencyMillis) +
check(unhealthyUpstream, maxEndpointLatencyMillis);
+ }
+
+ private int check(final Map<String, List<Upstream>> map, final AtomicLong
maxEndpointLatencyMillis) {
+ int submitted = 0;
for (Map.Entry<String, List<Upstream>> entry : map.entrySet()) {
String key = entry.getKey();
List<Upstream> value = entry.getValue();
for (Upstream upstream : value) {
- CompletableFuture<UpstreamWithSelectorId> future =
CompletableFuture.supplyAsync(() -> check(key, upstream), executor);
+ CompletableFuture<UpstreamWithSelectorId> future =
+ CompletableFuture.supplyAsync(() -> check(key,
upstream, maxEndpointLatencyMillis), executor);
futures.add(future);
+ submitted++;
}
}
+ return submitted;
}
- private UpstreamWithSelectorId check(final String selectorId, final
Upstream upstream) {
+ private UpstreamWithSelectorId check(final String selectorId, final
Upstream upstream,
+ final AtomicLong
maxEndpointLatencyMillis) {
if (!upstream.isHealthCheckEnabled()) {
if (!upstream.isHealthy()) {
upstream.setHealthy(true);
@@ -198,7 +292,15 @@ public final class UpstreamCheckTask implements Runnable {
}
return new UpstreamWithSelectorId(selectorId, upstream);
}
- boolean pass = UpstreamCheckUtils.checkUrl(upstream.getUrl(),
checkTimeout);
+ final long probeStart = System.nanoTime();
+ boolean pass;
+ try {
+ pass = UpstreamCheckUtils.checkUrl(upstream.getUrl(),
checkTimeout);
+ } finally {
+ final long latencyMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - probeStart);
+ maxEndpointLatencyMillis.accumulateAndGet(latencyMillis,
Math::max);
+ totalEndpointLatencyMillis.addAndGet(latencyMillis);
+ }
if (pass) {
if (upstream.isHealthy()) {
upstream.setLastHealthTimestamp(System.currentTimeMillis());
@@ -374,4 +476,99 @@ public final class UpstreamCheckTask implements Runnable {
public Map<String, List<Upstream>> getUnhealthyUpstream() {
return unhealthyUpstream;
}
+
+ /**
+ * Duration of the most recent health-check round in milliseconds,
measured with
+ * {@link System#nanoTime()} around the whole synchronized section (submit
+ wait).
+ * Zero until the first round completes.
+ *
+ * <p>Note: a round that could not start (because another was still in
flight) is still recorded,
+ * so a {@code 0} duration together with {@code 0} from {@link
#getLastRoundSubmittedChecks()}
+ * means "this round was skipped", not "the round was instant".
+ *
+ * @return last round duration in millis
+ */
+ public long getLastRoundDurationMillis() {
+ return lastRoundDurationMillis;
+ }
+
+ /**
+ * Number of endpoint probes submitted to the async executor during the
most recent round.
+ * A {@code 0} value alongside a {@code 0} {@link
#getLastRoundDurationMillis()} indicates the
+ * round was skipped because another round was already in flight, rather
than an empty population.
+ *
+ * @return submitted probe count
+ */
+ public int getLastRoundSubmittedChecks() {
+ return lastRoundSubmittedChecks;
+ }
+
+ /**
+ * Number of submitted probes that were still pending when the round
entered {@code waitFinish()}.
+ * A persistently high value indicates the executor is the bottleneck, not
the endpoints.
+ *
+ * @return pending probe count at wait time
+ */
+ public int getLastRoundPendingChecks() {
+ return lastRoundPendingChecks;
+ }
+
+ /**
+ * Maximum single-endpoint probe latency observed during the most recent
round, in milliseconds.
+ *
+ * @return max endpoint latency in millis
+ */
+ public long getLastRoundMaxEndpointLatencyMillis() {
+ return lastRoundMaxEndpointLatencyMillis;
+ }
+
+ /**
+ * Age of the oldest health-state timestamp across healthy and unhealthy
buckets, in milliseconds.
+ * Grows when rounds fall behind the configured interval. Upstreams that
have never been measured
+ * (unset {@code 0} timestamp, e.g. newly registered or
health-check-disabled while already healthy)
+ * are skipped, so an all-unmeasured population reports {@code 0} rather
than an epoch-sized age.
+ *
+ * @return oldest health-state age in millis
+ */
+ public long getLastRoundOldestHealthStateMillis() {
+ return lastRoundOldestHealthStateMillis;
+ }
+
+ /**
+ * Wall-clock timestamp at which the most recent round finished, in
milliseconds since epoch.
+ *
+ * @return last round completion timestamp
+ */
+ public long getLastRoundCompletedAtMillis() {
+ return lastRoundCompletedAtMillis;
+ }
+
+ /**
+ * Total number of health-check rounds completed since this task was
created.
+ *
+ * @return total rounds
+ */
+ public long getTotalRounds() {
+ return totalRounds.get();
+ }
+
+ /**
+ * Total number of endpoint probes submitted since this task was created.
+ *
+ * @return total submitted probes
+ */
+ public long getTotalSubmittedChecks() {
+ return totalSubmittedChecks.get();
+ }
+
+ /**
+ * Cumulative probe latency in milliseconds since this task was created.
Divide by
+ * {@link #getTotalSubmittedChecks()} for an average; only probes with
health-check
+ * enabled contribute.
+ *
+ * @return cumulative endpoint latency in millis
+ */
+ public long getTotalEndpointLatencyMillis() {
+ return totalEndpointLatencyMillis.get();
+ }
}
diff --git
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTaskTest.java
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTaskTest.java
index 8532f4e433..7dde55b878 100644
---
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTaskTest.java
+++
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTaskTest.java
@@ -303,4 +303,156 @@ public class UpstreamCheckTaskTest {
// Clean up
healthCheckTask.triggerRemoveAll(selectorId);
}
+
+ /**
+ * The observability counters added for #7647 must be populated after a
synchronous round
+ * completes, so operators can compare round duration, submitted probes
and pending probes
+ * against the configured checkInterval without adding a profiler to the
runtime.
+ */
+ @Test
+ @Timeout(30000)
+ public void roundMetricsAreRecordedAfterRunCompletes() {
+ final UpstreamCheckTask task = new UpstreamCheckTask(50000);
+ task.setPoolSize(2);
+ task.setCheckTimeout(500);
+ task.schedule();
+
+ final String selectorId = "metrics-round";
+ for (int i = 0; i < 4; i++) {
+ final Upstream upstream = Upstream.builder()
+ .url("127.0.0.1:" + (65000 + i))
+ .build();
+ upstream.setHealthy(true);
+ upstream.setLastHealthTimestamp(System.currentTimeMillis());
+ task.triggerAddOne(selectorId, upstream);
+ }
+
+ final long before = task.getTotalRounds();
+ task.run();
+
+ assertTrue(task.getTotalRounds() > before, "totalRounds must advance
after run()");
+ assertTrue(task.getLastRoundDurationMillis() >= 0L, "duration must be
non-negative");
+ assertTrue(task.getLastRoundSubmittedChecks() >= 4,
+ "submitted probes must cover every registered upstream, got "
+ task.getLastRoundSubmittedChecks());
+ assertTrue(task.getLastRoundPendingChecks() >= 0, "pending probes must
be non-negative");
+ assertTrue(task.getLastRoundCompletedAtMillis() > 0L, "completion
timestamp must be set");
+ assertTrue(task.getTotalSubmittedChecks() >=
task.getLastRoundSubmittedChecks(),
+ "cumulative submitted must include the last round");
+ assertTrue(task.getLastRoundOldestHealthStateMillis() >= 0L,
+ "oldest health-state age must be non-negative");
+
+ task.triggerRemoveAll(selectorId);
+ }
+
+ /**
+ * An empty population still completes a round and updates the round
counter, so operators
+ * can distinguish "task not running" from "task running with nothing to
check".
+ */
+ @Test
+ @Timeout(10000)
+ public void emptyPopulationStillRecordsRoundCompletion() {
+ final UpstreamCheckTask task = new UpstreamCheckTask(50000);
+ task.setPoolSize(1);
+ task.schedule();
+
+ final long before = task.getTotalRounds();
+ task.run();
+
+ assertTrue(task.getTotalRounds() > before, "round counter must advance
even with no upstreams");
+ assertTrue(task.getLastRoundDurationMillis() >= 0L, "duration must be
non-negative");
+ assertTrue(task.getLastRoundSubmittedChecks() == 0, "no probes should
be submitted for an empty population");
+ assertTrue(task.getLastRoundPendingChecks() == 0, "no probes should be
pending for an empty population");
+ assertTrue(task.getLastRoundMaxEndpointLatencyMillis() == 0L,
+ "endpoint latency must stay zero when no probe ran");
+ }
+
+ /**
+ * Upstreams registered with a stale {@code lastHealthTimestamp} must
surface through
+ * {@code lastRoundOldestHealthStateMillis}, so the metric actually
reflects health-state age
+ * instead of collapsing to zero.
+ */
+ @Test
+ @Timeout(10000)
+ public void oldestHealthStateAgeReflectsStaleTimestamp() {
+ final UpstreamCheckTask task = new UpstreamCheckTask(50000);
+ task.setPoolSize(1);
+ task.schedule();
+
+ final String selectorId = "metrics-stale";
+ // Use a blank URL so checkUrl() returns false without any network
I/O. A fixed port such as
+ // 127.0.0.1:65100 would let an unrelated listener flip the probe to
pass and refresh the
+ // timestamp, making the stale-age assertion flaky.
+ final Upstream stale = Upstream.builder().url("").build();
+ stale.setHealthy(true);
+ // Pretend the last successful probe happened 5 seconds ago.
+ stale.setLastHealthTimestamp(System.currentTimeMillis() - 5_000L);
+ task.triggerAddOne(selectorId, stale);
+
+ task.run();
+
+ assertTrue(task.getLastRoundOldestHealthStateMillis() >= 4_000L,
+ "oldest health-state age must reflect the stale timestamp, got
"
+ + task.getLastRoundOldestHealthStateMillis());
+
+ task.triggerRemoveAll(selectorId);
+ }
+
+ /**
+ * A health-check-disabled upstream that is already healthy never gets its
{@code lastHealthTimestamp}
+ * written, so it stays at the primitive default {@code 0}. Without a
guard the oldest-health-state
+ * metric would report {@code now - 0}, an epoch-sized (~56 years) age
that hides every real
+ * observation. The metric must skip such never-measured entries and
report a small age instead.
+ */
+ @Test
+ @Timeout(10000)
+ public void oldestHealthStateAgeSkipsNeverMeasuredUpstream() {
+ final UpstreamCheckTask task = new UpstreamCheckTask(50000);
+ task.setPoolSize(1);
+ task.schedule();
+
+ final String selectorId = "metrics-disabled";
+ final Upstream disabled =
Upstream.builder().url("").healthCheckEnabled(false).build();
+ // Healthy + disabled short-circuits the probe in check() without ever
writing a timestamp.
+ disabled.setHealthy(true);
+ task.triggerAddOne(selectorId, disabled);
+
+ task.run();
+
+ final long age = task.getLastRoundOldestHealthStateMillis();
+ assertTrue(age < TimeUnit.MINUTES.toMillis(1),
+ "never-measured upstream must be skipped, not reported as an
epoch-sized age, got " + age);
+
+ task.triggerRemoveAll(selectorId);
+ }
+
+ /**
+ * Rounds must be re-entrant safe: a second {@code run()} invocation
resets the per-round
+ * scratch counters so the exposed metrics never leak values from the
previous round.
+ */
+ @Test
+ @Timeout(30000)
+ public void perRoundCountersAreResetBetweenRounds() {
+ final UpstreamCheckTask task = new UpstreamCheckTask(50000);
+ task.setPoolSize(2);
+ task.setCheckTimeout(500);
+ task.schedule();
+
+ final String selectorId = "metrics-reset";
+ final Upstream upstream =
Upstream.builder().url("127.0.0.1:65200").build();
+ upstream.setHealthy(true);
+ upstream.setLastHealthTimestamp(System.currentTimeMillis());
+ task.triggerAddOne(selectorId, upstream);
+
+ task.run();
+ final int firstSubmitted = task.getLastRoundSubmittedChecks();
+ final long firstRounds = task.getTotalRounds();
+ assertTrue(firstSubmitted >= 1, "first round must submit at least one
probe");
+
+ task.triggerRemoveAll(selectorId);
+ task.run();
+
+ assertTrue(task.getTotalRounds() > firstRounds, "round counter must
advance across runs");
+ assertTrue(task.getLastRoundSubmittedChecks() == 0,
+ "second round must not carry over submitted count from the
first round");
+ }
}
