Add circuit-breaking for gRPC server and relative docs.

Author: wu-sheng

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.

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.

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"

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.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 @@ TTLStatusQuery.class, new TTLStatusQuery(
     public void start() throws ModuleStartException {
         grpcServer.addHandler(new RemoteServiceHandler(getManager()));
         grpcServer.addHandler(new HealthCheckServiceHandler());
+        grpcServer.addInterceptor(WatermarkGRPCInterceptor.INSTANCE);
 
         endpointNameGrouping.startHttpUriRecognitionSvr(
             getManager()

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/server/GRPCHandlerRegisterImpl.java

@@ -43,6 +43,6 @@ public void addHandler(ServerServiceDefinition definition) {
 
     @Override
     public void addFilter(ServerInterceptor interceptor) {
-        server.addHandler(interceptor);
+        server.addInterceptor(interceptor);
     }
 }

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

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() {

oap-server/server-library/library-server/src/main/java/org/apache/skywalking/oap/server/library/server/grpc/GRPCServer.java

@@ -139,7 +139,7 @@ public void addHandler(ServerServiceDefinition definition) {
         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);
     }

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.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 void start() throws ServiceNotProvidedException, ModuleStartException {
         service.addHandler(handler);
         service.addHandler(new AccessLogServiceGRPCHandlerV3(handler));
         service.addHandler(new SatelliteAccessLogServiceGRPCHandlerV3(handler));
+        service.addFilter(WatermarkGRPCInterceptor.INSTANCE);
     }
 
     @Override

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.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 void start() throws ServiceNotProvidedException, ModuleStartException {
         service.addHandler(new EBPFProfilingServiceHandler(getManager()));
         service.addHandler(new ContinuousProfilingServiceHandler(getManager(), this.config));
         service.addHandler(new AccessLogServiceHandler(getManager()));
+        service.addFilter(WatermarkGRPCInterceptor.INSTANCE);
     }
 
     @Override