You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2025/01/15 06:53:10 UTC
(skywalking) branch fix created (now 0c8412a504)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch fix
in repository https://gitbox.apache.org/repos/asf/skywalking.git
at 0c8412a504 1. Fix too many `RESOURCE EXHAUSTED: Watermark exceeded` when other nodes reached limitations. 2. Fix bug of watermark exceeded recovery.
This branch includes the following new commits:
new 0c8412a504 1. Fix too many `RESOURCE EXHAUSTED: Watermark exceeded` when other nodes reached limitations. 2. Fix bug of watermark exceeded recovery.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking) 01/01: 1. Fix too many `RESOURCE EXHAUSTED: Watermark exceeded` when other nodes reached limitations. 2. Fix bug of watermark exceeded recovery.
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch fix
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 0c8412a5049d3e3c2654d086f14c7eea9ede0a77
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Wed Jan 15 14:53:00 2025 +0800
1. Fix too many `RESOURCE EXHAUSTED: Watermark exceeded` when other nodes reached limitations.
2. Fix bug of watermark exceeded recovery.
---
.../server/core/remote/client/GRPCRemoteClient.java | 19 +++++++++++++++----
.../oap/server/core/watermark/WatermarkListener.java | 18 ++++++++++--------
2 files changed, 25 insertions(+), 12 deletions(-)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
index a1a324309f..0bfc460930 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
@@ -57,6 +57,7 @@ public class GRPCRemoteClient implements RemoteClient {
private CounterMetrics remoteOutCounter;
private CounterMetrics remoteOutErrorCounter;
private int remoteTimeout;
+ private long lastRemoteResourceExhaustedTime = 0;
public GRPCRemoteClient(final ModuleDefineHolder moduleDefineHolder,
final Address address,
@@ -79,7 +80,8 @@ public class GRPCRemoteClient implements RemoteClient {
"The number(client side) of inside remote inside aggregate rpc.",
new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(
address
- .toString(), "N")
+ .toString(), "N"
+ )
);
remoteOutErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
@@ -89,7 +91,8 @@ public class GRPCRemoteClient implements RemoteClient {
"The error number(client side) of inside remote inside aggregate rpc.",
new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(
address
- .toString(), "N")
+ .toString(), "N"
+ )
);
}
@@ -196,8 +199,9 @@ public class GRPCRemoteClient implements RemoteClient {
sleepTotalMillis += sleepMillis;
if (sleepTotalMillis > 60000) {
- log.warn("Remote client [{}] block times over 60 seconds. Current streaming number {}",
- address, concurrentStreamObserverNumber.get()
+ log.warn(
+ "Remote client [{}] block times over 60 seconds. Current streaming number {}",
+ address, concurrentStreamObserverNumber.get()
);
// Reset sleepTotalMillis to avoid too many warn logs.
sleepTotalMillis = 0;
@@ -220,6 +224,13 @@ public class GRPCRemoteClient implements RemoteClient {
log.debug(throwable.getMessage(), throwable);
}
return;
+ } else if (Status.RESOURCE_EXHAUSTED.getCode() == status.getCode()) {
+ if (System.currentTimeMillis() - lastRemoteResourceExhaustedTime > 120_000) {
+ // Only output the log every 120 seconds.
+ log.warn(throwable.getMessage(), throwable);
+ lastRemoteResourceExhaustedTime = System.currentTimeMillis();
+ }
+ return;
}
log.error(throwable.getMessage(), throwable);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
index 0eb5c87d0a..295f232db8 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
@@ -24,12 +24,12 @@ import lombok.Getter;
/**
* WatermarkListener is the listener for receiving WatermarkEvent and react to it.
* The implementations of this listener has two ways to interact with the WatermarkEvent:
- * 1. use {@link #isWatermarkExceeded()} to check if the watermark is exceeded.
+ * 1. use {@link #isWatermarkExceeded()} == true to check if the watermark is exceeded.
* 2. override {@link #beAwareOf(WatermarkEvent.Type)} to react to the event.
*
* When the oap recovered from the limiting state, the listener has two ways to be aware of it:
- * 1. use {@link #isHealthy()} to check if the watermark is recovered.
- * 2. Be notified by calling {@link #beAwareOfRecovery()}.
+ * 1. use {@link #isWatermarkExceeded()} == false to check if the watermark is recovered.
+ * 2. Be notified by calling {@link #beAwareOfRecovery()}}.
*/
public abstract class WatermarkListener {
@Getter
@@ -58,18 +58,20 @@ public abstract class WatermarkListener {
return false;
}
- void isHealthy() {
- isWatermarkExceeded = false;
- }
-
public boolean isWatermarkExceeded() {
return isWatermarkExceeded;
}
+ /**
+ * Receive the WatermarkEvent and react to it.
+ */
protected void beAwareOf(WatermarkEvent.Type event) {
}
+ /**
+ * Receive the recovery status and react to it.
+ */
protected void beAwareOfRecovery() {
- // Do nothing by default.
+ isWatermarkExceeded = false;
}
}