Copilot commented on code in PR #7674:
URL: https://github.com/apache/shenyu/pull/7674#discussion_r4291647693
##########
shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java:
##########
@@ -156,6 +185,11 @@ public void run() {
}
private void healthCheck() {
+ final long startNanos = System.nanoTime();
+ currentRoundSubmittedChecks.set(0);
+ currentRoundMaxEndpointLatencyMillis.set(0L);
Review Comment:
A concurrent run() resets these shared counters before acquiring lock, so it
can erase the active round's submitted count or maximum latency. Recording
metrics after releasing lock also lets the next invocation reset the maximum
before it is captured. Isolate counters per round and pass that context to its
probes, or serialize reset, capture and publication while accounting for probes
still running after an exception.
##########
shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java:
##########
@@ -164,14 +198,70 @@ private void healthCheck() {
synchronized (lock) {
if (tryStartHealthCheck()) {
doHealthCheck();
+ submitted = currentRoundSubmittedChecks.get();
+ pendingAtWait = countPendingFutures();
waitFinish();
}
}
} catch (Exception e) {
LOG.error("[Health Check] Meet problem: ", e);
} finally {
finishHealthCheck();
+ recordRoundMetrics(startNanos, submitted, pendingAtWait);
+ }
+ }
+
+ 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 durationMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ final long maxEndpointLatencyMillis =
currentRoundMaxEndpointLatencyMillis.get();
+ final long oldestHealthStateMillis = computeOldestHealthStateMillis();
+ final long completedAtMillis = System.currentTimeMillis();
+ this.lastRoundDurationMillis = durationMillis;
+ this.lastRoundSubmittedChecks = submitted;
+ this.lastRoundPendingChecks = pendingAtWait;
+ this.lastRoundMaxEndpointLatencyMillis = maxEndpointLatencyMillis;
+ this.lastRoundOldestHealthStateMillis = oldestHealthStateMillis;
Review Comment:
Separate volatile writes do not publish the coherent snapshot promised above
the scratch counters. A monitoring thread can read the new duration while
submitted/pending counts, latency or completion time still belong to the
previous round. Publish an immutable round-metrics object through one volatile
reference and expose a snapshot accessor so consumers can read one consistent
round; separate scalar getters alone cannot guarantee that.
##########
shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java:
##########
@@ -164,14 +198,70 @@ private void healthCheck() {
synchronized (lock) {
if (tryStartHealthCheck()) {
doHealthCheck();
+ submitted = currentRoundSubmittedChecks.get();
+ pendingAtWait = countPendingFutures();
waitFinish();
}
}
} catch (Exception e) {
LOG.error("[Health Check] Meet problem: ", e);
} finally {
finishHealthCheck();
+ recordRoundMetrics(startNanos, submitted, pendingAtWait);
+ }
+ }
+
+ 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 durationMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ final long maxEndpointLatencyMillis =
currentRoundMaxEndpointLatencyMillis.get();
+ 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 long computeOldestHealthStateMillis() {
+ final long now = System.currentTimeMillis();
+ long oldest = 0L;
+ for (List<Upstream> bucket : healthyUpstream.values()) {
+ for (Upstream upstream : bucket) {
+ final long age = now - upstream.getLastHealthTimestamp();
+ if (age > oldest) {
+ oldest = age;
+ }
+ }
+ }
+ for (List<Upstream> bucket : unhealthyUpstream.values()) {
+ for (Upstream upstream : bucket) {
+ final long age = now - upstream.getLastUnhealthyTimestamp();
+ if (age > oldest) {
+ oldest = age;
+ }
+ }
+ }
Review Comment:
Health-check-disabled upstreams can permanently dominate this metric with an
age since 1970. UpstreamCacheManager.initializeUpstreamHealthStatus() marks
them healthy without setting lastHealthTimestamp, and check() leaves that zero
timestamp unchanged while they remain healthy. Exclude disabled entries and
unset timestamps in both buckets so they do not hide the ages of actual health
observations.
##########
shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTaskTest.java:
##########
@@ -303,4 +303,125 @@ public void testMoveUpstreamBetweenMaps() {
// 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";
+ final Upstream stale =
Upstream.builder().url("127.0.0.1:65100").build();
Review Comment:
This test assumes port 65100 has no listener. If another process is
listening, checkUrl() succeeds and refreshes lastHealthTimestamp, causing the
stale-age assertion to fail. A blank URL returns false without network I/O and
preserves the stale timestamp under this test's configured threshold.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
