server: ensure errors not logged when client closes connection (#84)

Errors were being logged when client Envoy closes connection while being shut down/restarted.
This change ensures that errors are not logged when the error code is CANCELLED.

Fixes #83

Signed-off-by: Vaidotas Valuckas <[email protected]>

Author: rieske

server/src/main/java/io/envoyproxy/controlplane/server/DiscoveryServer.java

@@ -221,7 +221,7 @@ public void onNext(DiscoveryRequest request) {
 
       @Override
       public void onError(Throwable t) {
-        if (!Status.fromThrowable(t).equals(Status.CANCELLED)) {
+        if (!Status.fromThrowable(t).getCode().equals(Status.CANCELLED.getCode())) {
           LOGGER.error("[{}] stream closed with error", streamId, t);
         }
 

server/src/test/java/io/envoyproxy/controlplane/server/DiscoveryServerTest.java

@@ -37,8 +37,12 @@
 import io.envoyproxy.controlplane.cache.Watch;
 import io.envoyproxy.controlplane.cache.WatchCancelledException;
 import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import io.grpc.testing.GrpcServerRule;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -748,6 +752,58 @@ public void onStreamCloseWithError(long streamId, String typeUrl, Throwable erro
     assertThat(callbacks.streamResponseCount).hasValue(0);
   }
 
+  @Test
+  public void callbackOnError_logsError_onException() {
+    MockConfigWatcher configWatcher = new MockConfigWatcher(false, createResponses());
+    DiscoveryServer server = new DiscoveryServer(configWatcher);
+
+    AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase service =
+        server.getAggregatedDiscoveryServiceImpl();
+
+    MockDiscoveryResponseObserver responseObserver = new MockDiscoveryResponseObserver();
+    StreamObserver<DiscoveryRequest> requestObserver = service.streamAggregatedResources(responseObserver);
+
+    try {
+      ByteArrayOutputStream stdErr = new ByteArrayOutputStream();
+      System.setErr(new PrintStream(stdErr));
+
+      requestObserver.onError(new StatusRuntimeException(Status.INTERNAL
+          .withDescription("internal error")
+          .withCause(new RuntimeException("some error"))));
+
+      assertThat(stdErr.toString()).contains("ERROR ");
+      assertThat(stdErr.toString()).contains("io.grpc.StatusRuntimeException: INTERNAL: internal error");
+    } finally {
+      System.setErr(System.err);
+    }
+  }
+
+  @Test
+  public void callbackOnError_doesNotLogError_whenCancelled() {
+    MockConfigWatcher configWatcher = new MockConfigWatcher(false, createResponses());
+    DiscoveryServer server = new DiscoveryServer(configWatcher);
+
+    AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase service =
+        server.getAggregatedDiscoveryServiceImpl();
+
+    MockDiscoveryResponseObserver responseObserver = new MockDiscoveryResponseObserver();
+    StreamObserver<DiscoveryRequest> requestObserver = service.streamAggregatedResources(responseObserver);
+
+    try {
+      ByteArrayOutputStream stdErr = new ByteArrayOutputStream();
+      System.setErr(new PrintStream(stdErr));
+
+      requestObserver.onError(new StatusRuntimeException(Status.CANCELLED
+          .withDescription("internal error")
+          .withCause(new RuntimeException("some error"))));
+
+      assertThat(stdErr.toString()).doesNotContain("ERROR ");
+      assertThat(stdErr.toString()).doesNotContain("io.grpc.StatusRuntimeException: CANCELLED:");
+    } finally {
+      System.setErr(System.err);
+    }
+  }
+
   private static Table<String, String, Collection<? extends Message>> createResponses() {
     return ImmutableTable.<String, String, Collection<? extends Message>>builder()
         .put(Resources.CLUSTER_TYPE_URL, VERSION, ImmutableList.of(CLUSTER))