core: Count client-cancelled server calls as failed in Channelz

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -46,6 +46,7 @@
 import io.perfmark.Tag;
 import io.perfmark.TaskCloseable;
 import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -66,6 +67,8 @@ final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
   private final DecompressorRegistry decompressorRegistry;
   private final CompressorRegistry compressorRegistry;
   private CallTracer serverCallTracer;
+  // The application's close() and the transport's closed() may race; only the first is counted.
+  private final AtomicBoolean callEndReported = new AtomicBoolean();
 
   // state
   private volatile boolean cancelled;
@@ -229,7 +232,13 @@ private void closeInternal(Status status, Metadata trailers) {
 
       stream.close(status, trailers);
     } finally {
-      serverCallTracer.reportCallEnded(status.isOk());
+      reportCallEnded(status.isOk());
+    }
+  }
+
+  private void reportCallEnded(boolean success) {
+    if (callEndReported.compareAndSet(false, true)) {
+      serverCallTracer.reportCallEnded(success);
     }
   }
 
@@ -287,7 +296,7 @@ private void handleInternalError(Throwable internalError) {
         : Status.INTERNAL.withCause(internalError)
             .withDescription("Internal error so cancelling stream.");
     stream.cancel(status);
-    serverCallTracer.reportCallEnded(false); // error so always false
+    reportCallEnded(false); // error so always false
   }
 
   /**
@@ -415,6 +424,7 @@ private void closedInternal(Status status) {
           listener.onComplete();
         } else {
           call.cancelled = true;
+          call.reportCallEnded(false);
           listener.onCancel();
           // The status will not have a cause in all failure scenarios, but we want to make sure
           // we always cancel the context with one to keep the context cancelled state consistent.

core/src/test/java/io/grpc/internal/ServerCallImplTest.java

@@ -127,6 +127,57 @@ public void callTracer_failure() {
     callTracer0(Status.UNKNOWN);
   }
 
+  @Test
+  public void callTracer_clientCancelled_reportsCallFailed() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+
+    streamListener.closed(Status.CANCELLED);
+
+    ServerStats.Builder afterBuilder = new ServerStats.Builder();
+    serverCallTracer.updateBuilder(afterBuilder);
+    ServerStats after = afterBuilder.build();
+    assertEquals(1, after.callsStarted);
+    assertEquals(0, after.callsSucceeded);
+    assertEquals(1, after.callsFailed);
+  }
+
+  @Test
+  public void callTracer_clientCancelledThenClosed_reportsCallFailedOnce() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+
+    streamListener.closed(Status.CANCELLED);
+    call.sendHeaders(new Metadata());
+    call.sendMessage(123L);
+    call.close(Status.OK, new Metadata());
+
+    ServerStats.Builder afterBuilder = new ServerStats.Builder();
+    serverCallTracer.updateBuilder(afterBuilder);
+    ServerStats after = afterBuilder.build();
+    assertEquals(1, after.callsStarted);
+    assertEquals(0, after.callsSucceeded);
+    assertEquals(1, after.callsFailed);
+  }
+
+  @Test
+  public void callTracer_closedThenClientCancelled_reportsCallSucceededOnce() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+
+    call.sendHeaders(new Metadata());
+    call.sendMessage(123L);
+    call.close(Status.OK, new Metadata());
+    streamListener.closed(Status.CANCELLED);
+
+    ServerStats.Builder afterBuilder = new ServerStats.Builder();
+    serverCallTracer.updateBuilder(afterBuilder);
+    ServerStats after = afterBuilder.build();
+    assertEquals(1, after.callsStarted);
+    assertEquals(1, after.callsSucceeded);
+    assertEquals(0, after.callsFailed);
+  }
+
   private void callTracer0(Status status) {
     CallTracer tracer = CallTracer.getDefaultFactory().create();
     ServerStats.Builder beforeBuilder = new ServerStats.Builder();