Do not log grpc envoy receiver client-side streaming cancellation as errors (#9843)

docs/en/changes/changes.md

@@ -69,6 +69,8 @@
 * [**Breaking Change**] Support new records query protocol, rename the column named `service_id` to `entity_id` for support difference entity.
   Please re-create `top_n_database_statement` index/table.
 * Remove improper self-obs metrics in JvmMetricsHandler(for Kafka channel).
+* gRPC stream canceling code is not logged as an error when the client cancels the stream. The client 
+  cancels the stream when the pod is terminated.
 
 #### UI
 

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

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.core.remote;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import java.util.Objects;
 import org.apache.skywalking.oap.server.core.CoreModule;
@@ -141,6 +142,13 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (LOGGER.isDebugEnabled()) {
+                        LOGGER.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 LOGGER.error(throwable.getMessage(), throwable);
             }
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.core.remote.client;
 
 import io.grpc.ManagedChannel;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.netty.handler.ssl.SslContext;
 import java.util.List;
@@ -222,6 +223,13 @@ public class GRPCRemoteClient implements RemoteClient {
                            @Override
                            public void onError(Throwable throwable) {
                                concurrentStreamObserverNumber.addAndGet(-1);
+                               Status status = Status.fromThrowable(throwable);
+                               if (Status.CANCELLED.getCode() == status.getCode()) {
+                                   if (log.isDebugEnabled()) {
+                                       log.debug(throwable.getMessage(), throwable);
+                                   }
+                                   return;
+                               }
                                log.error(throwable.getMessage(), throwable);
                            }
 

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java

@@ -23,6 +23,7 @@ import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
 import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
 import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
 import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import java.util.ArrayList;
 import java.util.List;
@@ -170,6 +171,13 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (LOGGER.isDebugEnabled()) {
+                        LOGGER.debug("Envoy client cancelled sending access logs", throwable);
+                    }
+                    return;
+                }
                 LOGGER.error("Error in receiving access log from envoy", throwable);
             }
 

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/MetricServiceGRPCHandler.java

@@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.receiver.envoy;
 import io.envoyproxy.envoy.service.metrics.v2.MetricsServiceGrpc;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsMessage;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsResponse;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.prometheus.client.Metrics;
 import java.util.ArrayList;
@@ -129,6 +130,13 @@ public class MetricServiceGRPCHandler extends MetricsServiceGrpc.MetricsServiceI
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.error("Envoy client cancelled sending metrics", throwable);
+                    }
+                    return;
+                }
                 log.error("Error in receiving metrics from envoy", throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/EBPFProfilingServiceHandler.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.receiver.ebpf.provider.handler;
 
 import com.google.common.base.Joiner;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
@@ -187,6 +188,13 @@ public class EBPFProfilingServiceHandler extends EBPFProfilingServiceGrpc.EBPFPr
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 log.error("Error in receiving ebpf profiling data", throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-event-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/event/grpc/EventGrpcServiceHandler.java

@@ -93,6 +93,13 @@ public class EventGrpcServiceHandler extends EventServiceGrpc.EventServiceImplBa
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 log.error(throwable.getMessage(), throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.aop.server.receiver.mesh;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import org.apache.skywalking.apm.network.servicemesh.v3.MeshProbeDownstream;
 import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
@@ -47,6 +48,13 @@ public class MeshGRPCHandler extends ServiceMeshMetricServiceGrpc.ServiceMeshMet
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (LOGGER.isDebugEnabled()) {
+                        LOGGER.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 LOGGER.error(throwable.getMessage(), throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-meter-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/meter/provider/handler/MeterServiceHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.meter.provider.handler;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.network.common.v3.Commands;
@@ -76,6 +77,13 @@ public class MeterServiceHandler extends MeterReportServiceGrpc.MeterReportServi
             @Override
             public void onError(Throwable throwable) {
                 processor.process();
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 log.error(throwable.getMessage(), throwable);
             }
 
@@ -105,6 +113,13 @@ public class MeterServiceHandler extends MeterReportServiceGrpc.MeterReportServi
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 log.error(throwable.getMessage(), throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-profile-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/profile/provider/handler/ProfileTaskServiceHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.profile.provider.handler;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -111,6 +112,13 @@ public class ProfileTaskServiceHandler extends ProfileTaskGrpc.ProfileTaskImplBa
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (LOGGER.isDebugEnabled()) {
+                        LOGGER.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 LOGGER.error(throwable.getMessage(), throwable);
             }
 

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/grpc/TraceSegmentReportServiceHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.grpc;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.network.common.v3.Commands;
@@ -80,6 +81,13 @@ public class TraceSegmentReportServiceHandler extends TraceSegmentReportServiceG
 
             @Override
             public void onError(Throwable throwable) {
+                Status status = Status.fromThrowable(throwable);
+                if (Status.CANCELLED.getCode() == status.getCode()) {
+                    if (log.isDebugEnabled()) {
+                        log.debug(throwable.getMessage(), throwable);
+                    }
+                    return;
+                }
                 log.error(throwable.getMessage(), throwable);
             }