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/10 10:00:00 UTC
(skywalking) branch watermark updated: Add circuit-breaking for gRPC server and relative docs.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch watermark
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/watermark by this push:
new 39e4882193 Add circuit-breaking for gRPC server and relative docs.
39e4882193 is described below
commit 39e4882193b359f0bae106943ac608e816042307
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Jan 10 17:59:53 2025 +0800
Add circuit-breaking for gRPC server and relative docs.
---
docs/en/setup/backend/backend-telemetry.md | 6 +-
docs/en/setup/backend/circuit-breaking.md | 20 +++++++
docs/menu.yml | 2 +
.../oap/server/core/CoreModuleProvider.java | 2 +
.../core/server/GRPCHandlerRegisterImpl.java | 2 +-
.../core/watermark/WatermarkGRPCInterceptor.java | 67 ++++++++++++++++++++++
.../server/core/watermark/WatermarkWatcher.java | 8 ++-
.../oap/server/library/server/grpc/GRPCServer.java | 2 +-
.../envoy/EnvoyMetricReceiverProvider.java | 2 +
.../ebpf/provider/EBPFReceiverProvider.java | 2 +
.../server/SharingServerModuleProvider.java | 2 +
.../src/main/resources/application.yml | 4 ++
12 files changed, 112 insertions(+), 7 deletions(-)
diff --git a/docs/en/setup/backend/backend-telemetry.md b/docs/en/setup/backend/backend-telemetry.md
index 656063fcca..2ddd403b7d 100644
--- a/docs/en/setup/backend/backend-telemetry.md
+++ b/docs/en/setup/backend/backend-telemetry.md
@@ -1,11 +1,11 @@
# Telemetry for backend
The OAP backend cluster itself is a distributed streaming process system. To assist the Ops team, we provide the telemetry for the OAP backend itself, also known as self-observability (so11y)
-By default, the telemetry is disabled by setting `selector` to `none`, like this:
+By default, the telemetry is disabled by setting `selector` to `prometheus`, like this, which activated the Prometheus telemetry.
```yaml
telemetry:
- selector: ${SW_TELEMETRY:none}
+ selector: ${SW_TELEMETRY:prometheus}
none:
prometheus:
host: ${SW_TELEMETRY_PROMETHEUS_HOST:0.0.0.0}
@@ -15,7 +15,7 @@ telemetry:
sslCertChainPath: ${SW_TELEMETRY_PROMETHEUS_SSL_CERT_CHAIN_PATH:""}
```
-You may also set `Prometheus` to enable them. For more information, refer to the details below.
+Besides the self observability, this supports [OAP circuit breaking](circuit-breaker.md) functionality.
## Self Observability
SkyWalking supports exposing telemetry data representing OAP running status through Prometheus endpoint.
diff --git a/docs/en/setup/backend/circuit-breaking.md b/docs/en/setup/backend/circuit-breaking.md
new file mode 100644
index 0000000000..1f2d5e4037
--- /dev/null
+++ b/docs/en/setup/backend/circuit-breaking.md
@@ -0,0 +1,20 @@
+# Circuit Breaking
+
+Circuit breaking is a mechanism used to detect failures and encapsulate the logic of preventing OAP node crashing. It is
+a key component of SkyWalking's resilience strategy. This approach protects the system from overload and ensures
+stability.
+
+Currently, there are two available strategies for circuit breaking: heap memory usage and direct memory pool size.
+
+```yaml
+# The int value of the max heap memory usage percent. The default value is 85%.
+maxHeapMemoryUsagePercent: ${SW_CORE_MAX_HEAP_MEMORY_USAGE_PERCENT:85}
+# The long value of the max direct memory usage. The default max value is -1, representing no limit.
+maxDirectMemoryUsage: ${SW_CORE_MAX_DIRECT_MEMORY_USAGE:-1}
+```
+
+SkyWalking's circuit breaker mechanism actively rejects new telemetry data from gRPC.
+We keep HTTP services running to serve the UI and API, and Prometheus telemetry data is still available.
+
+Note, this feature relies on the `prometheus` provider in [Telemetry for backend](backend-telemetry.md) to monitor OAP
+server status.
\ No newline at end of file
diff --git a/docs/menu.yml b/docs/menu.yml
index 8bb5acd462..5944038efb 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -196,6 +196,8 @@ catalog:
path: "/en/setup/backend/backend-telemetry"
- name: "OAP Health Check"
path: "/en/setup/backend/backend-health-check"
+ - name: "Circuit Breaking"
+ path: "/en/setup/backend/circuit-breaking"
- name: "BanyanDB Exclusive Setup"
catalog:
- name: "Progressive TTL"
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
index edbca88f2d..dde5ee3f76 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
@@ -102,6 +102,7 @@ import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
import org.apache.skywalking.oap.server.core.storage.ttl.DataTTLKeeperTimer;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkGRPCInterceptor;
import org.apache.skywalking.oap.server.core.watermark.WatermarkWatcher;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
@@ -392,6 +393,7 @@ public class CoreModuleProvider extends ModuleProvider {
public void start() throws ModuleStartException {
grpcServer.addHandler(new RemoteServiceHandler(getManager()));
grpcServer.addHandler(new HealthCheckServiceHandler());
+ grpcServer.addInterceptor(WatermarkGRPCInterceptor.INSTANCE);
endpointNameGrouping.startHttpUriRecognitionSvr(
getManager()
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/server/GRPCHandlerRegisterImpl.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/server/GRPCHandlerRegisterImpl.java
index f75f35598a..65d5b97013 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/server/GRPCHandlerRegisterImpl.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/server/GRPCHandlerRegisterImpl.java
@@ -43,6 +43,6 @@ public class GRPCHandlerRegisterImpl implements GRPCHandlerRegister {
@Override
public void addFilter(ServerInterceptor interceptor) {
- server.addHandler(interceptor);
+ server.addInterceptor(interceptor);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkGRPCInterceptor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkGRPCInterceptor.java
new file mode 100644
index 0000000000..e3e500babc
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkGRPCInterceptor.java
@@ -0,0 +1,67 @@
+/*
+ * 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.skywalking.oap.server.core.watermark;
+
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+
+/**
+ * gRPCWatermarkInterceptor is a gRPC interceptor that checks if the watermark is exceeded before processing the request.
+ */
+public class WatermarkGRPCInterceptor extends WatermarkListener implements ServerInterceptor {
+ public static WatermarkGRPCInterceptor INSTANCE;
+
+ private WatermarkGRPCInterceptor() {
+ super("gRPC-Watermark-Interceptor");
+ }
+
+ public static WatermarkGRPCInterceptor create() {
+ INSTANCE = new WatermarkGRPCInterceptor();
+ return INSTANCE;
+ }
+
+ @Override
+ public <REQ, RESP> ServerCall.Listener<REQ> interceptCall(final ServerCall<REQ, RESP> call,
+ final Metadata headers,
+ final ServerCallHandler<REQ, RESP> next) {
+ if (isWatermarkExceeded()) {
+ call.close(Status.RESOURCE_EXHAUSTED.withDescription("Watermark exceeded"), new Metadata());
+ return new ServerCall.Listener<REQ>() {
+ };
+ }
+
+ ServerCall.Listener<REQ> delegate = next.startCall(call, headers);
+
+ return new ForwardingServerCallListener.SimpleForwardingServerCallListener<REQ>(delegate) {
+ @Override
+ public void onMessage(final REQ message) {
+ if (isWatermarkExceeded()) {
+ call.close(Status.RESOURCE_EXHAUSTED.withDescription("Watermark exceeded"), new Metadata());
+ return;
+ }
+
+ super.onMessage(message);
+ }
+ };
+ }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
index e58e89a130..23ca3a650c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
@@ -54,10 +54,14 @@ public class WatermarkWatcher implements Service {
public void start(MetricsCollector so11yCollector) {
this.so11yCollector = so11yCollector;
- Executors.newSingleThreadScheduledExecutor()
- .scheduleWithFixedDelay(this::watch, 0, 10, TimeUnit.SECONDS);
lock = new ReentrantLock();
listeners = new ArrayList<>();
+
+ // Create internal listeners to watch the watermark
+ this.addListener(WatermarkGRPCInterceptor.create());
+
+ Executors.newSingleThreadScheduledExecutor()
+ .scheduleWithFixedDelay(this::watch, 0, 10, TimeUnit.SECONDS);
}
private void watch() {
diff --git a/oap-server/server-library/library-server/src/main/java/org/apache/skywalking/oap/server/library/server/grpc/GRPCServer.java b/oap-server/server-library/library-server/src/main/java/org/apache/skywalking/oap/server/library/server/grpc/GRPCServer.java
index 67d6420008..068597b166 100644
--- a/oap-server/server-library/library-server/src/main/java/org/apache/skywalking/oap/server/library/server/grpc/GRPCServer.java
+++ b/oap-server/server-library/library-server/src/main/java/org/apache/skywalking/oap/server/library/server/grpc/GRPCServer.java
@@ -139,7 +139,7 @@ public class GRPCServer implements Server {
nettyServerBuilder.addService(definition);
}
- public void addHandler(ServerInterceptor serverInterceptor) {
+ public void addInterceptor(ServerInterceptor serverInterceptor) {
log.info("Bind interceptor {} into gRPC server {}:{}", serverInterceptor.getClass().getSimpleName(), host, port);
nettyServerBuilder.intercept(serverInterceptor);
}
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverProvider.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverProvider.java
index f5ce2dc96c..ac4bcd5c1e 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverProvider.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/EnvoyMetricReceiverProvider.java
@@ -25,6 +25,7 @@ import org.apache.skywalking.oap.server.core.RunningMode;
import org.apache.skywalking.oap.server.core.oal.rt.OALEngineLoaderService;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegisterImpl;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkGRPCInterceptor;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@@ -138,6 +139,7 @@ public class EnvoyMetricReceiverProvider extends ModuleProvider {
service.addHandler(handler);
service.addHandler(new AccessLogServiceGRPCHandlerV3(handler));
service.addHandler(new SatelliteAccessLogServiceGRPCHandlerV3(handler));
+ service.addFilter(WatermarkGRPCInterceptor.INSTANCE);
}
@Override
diff --git a/oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/EBPFReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/EBPFReceiverProvider.java
index cb78240b06..6041ac865b 100644
--- a/oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/EBPFReceiverProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/EBPFReceiverProvider.java
@@ -24,6 +24,7 @@ import org.apache.skywalking.oap.server.core.RunningMode;
import org.apache.skywalking.oap.server.core.oal.rt.OALEngineLoaderService;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegisterImpl;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkGRPCInterceptor;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@@ -122,6 +123,7 @@ public class EBPFReceiverProvider extends ModuleProvider {
service.addHandler(new EBPFProfilingServiceHandler(getManager()));
service.addHandler(new ContinuousProfilingServiceHandler(getManager(), this.config));
service.addHandler(new AccessLogServiceHandler(getManager()));
+ service.addFilter(WatermarkGRPCInterceptor.INSTANCE);
}
@Override
diff --git a/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/SharingServerModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/SharingServerModuleProvider.java
index bfd78ab7a4..f1c9c4c166 100644
--- a/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/SharingServerModuleProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/sharing/server/SharingServerModuleProvider.java
@@ -28,6 +28,7 @@ import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegisterImpl;
import org.apache.skywalking.oap.server.core.server.HTTPHandlerRegister;
import org.apache.skywalking.oap.server.core.server.HTTPHandlerRegisterImpl;
import org.apache.skywalking.oap.server.core.server.auth.AuthenticationInterceptor;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkGRPCInterceptor;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@@ -148,6 +149,7 @@ public class SharingServerModuleProvider extends ModuleProvider {
public void start() {
if (Objects.nonNull(grpcServer)) {
grpcServer.addHandler(new HealthCheckServiceHandler());
+ grpcServer.addInterceptor(WatermarkGRPCInterceptor.INSTANCE);
}
if (Objects.nonNull(receiverGRPCHandlerRegister)) {
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 6481774f66..b2bb3548ff 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -141,6 +141,10 @@ core:
# All the hierarchy relations are defined in the `hierarchy-definition.yml`.
# Notice: some of the configurations only available for kubernetes environments.
enableHierarchy: ${SW_CORE_ENABLE_HIERARCHY:true}
+ # The int value of the max heap memory usage percent. The default value is 85%.
+ maxHeapMemoryUsagePercent: ${SW_CORE_MAX_HEAP_MEMORY_USAGE_PERCENT:85}
+ # The long value of the max direct memory usage. The default max value is -1, representing no limit.
+ maxDirectMemoryUsage: ${SW_CORE_MAX_DIRECT_MEMORY_USAGE:-1}
storage:
selector: ${SW_STORAGE:banyandb}
banyandb: