core: Count 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,9 @@ final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
   private final DecompressorRegistry decompressorRegistry;
   private final CompressorRegistry compressorRegistry;
   private CallTracer serverCallTracer;
+  // Ensures the call end is reported to serverCallTracer exactly once. Both the application
+  // (via close()) and the transport (via ServerStreamListener.closed()) may report it.
+  private final AtomicBoolean callEndReported = new AtomicBoolean();
 
   // state
   private volatile boolean cancelled;
@@ -226,7 +230,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);
     }
   }
 
@@ -284,7 +294,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
   }
 
   /**
@@ -406,6 +416,9 @@ private void closedInternal(Status status) {
         GrpcUtil.closeQuietly(delayedMessage);
         delayedMessage = null;
       }
+      // The application may never call close() if the call was cancelled, e.g. by the client or
+      // the transport, so report the end of the call here too.
+      call.reportCallEnded(status.isOk());
       Throwable cancelCause = null;
       try {
         if (status.isOk()) {

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

@@ -156,6 +156,67 @@ private void callTracer0(Status status) {
     }
   }
 
+  @Test
+  public void callTracer_clientCancelled_reportsCallFailed() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+
+    streamListener.closed(Status.CANCELLED);
+
+    ServerStats after = getServerStats(serverCallTracer);
+    assertEquals(1, after.callsStarted);
+    assertEquals(0, after.callsSucceeded);
+    assertEquals(1, after.callsFailed);
+  }
+
+  @Test
+  public void callTracer_clientCancelledThenServerClosed_reportsOnlyOnce() {
+    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 after = getServerStats(serverCallTracer);
+    assertEquals(1, after.callsStarted);
+    assertEquals(0, after.callsSucceeded);
+    assertEquals(1, after.callsFailed);
+  }
+
+  @Test
+  public void callTracer_serverClosedThenStreamClosed_reportsOnlyOnce() {
+    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 after = getServerStats(serverCallTracer);
+    assertEquals(1, after.callsStarted);
+    assertEquals(1, after.callsSucceeded);
+    assertEquals(0, after.callsFailed);
+  }
+
+  @Test
+  public void callTracer_missingResponse_reportsOnlyFailure() {
+    call.close(Status.OK, new Metadata());
+
+    ServerStats after = getServerStats(serverCallTracer);
+    assertEquals(1, after.callsStarted);
+    assertEquals(0, after.callsSucceeded);
+    assertEquals(1, after.callsFailed);
+  }
+
+  private static ServerStats getServerStats(CallTracer tracer) {
+    ServerStats.Builder builder = new ServerStats.Builder();
+    tracer.updateBuilder(builder);
+    return builder.build();
+  }
+
   @Test
   public void request() {
     call.request(10);