opentelemetry: plumb subchannel metrics disconnect error (#12342)

Finishes the remaining work of
[A94](https://github.com/grpc/proposal/pull/485/files) i.e. the plumbing the disconnect error

Author: AgraVator

binder/src/androidTest/java/io/grpc/binder/internal/BinderClientTransportTest.java

@@ -49,6 +49,7 @@
 import io.grpc.internal.ClientStream;
 import io.grpc.internal.ClientStreamListener;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
+import io.grpc.internal.DisconnectError;
 import io.grpc.internal.FixedObjectPool;
 import io.grpc.internal.ManagedClientTransport;
 import io.grpc.internal.ObjectPool;
@@ -529,7 +530,7 @@ private static final class TestTransportListener implements ManagedClientTranspo
     private final SettableFuture<Boolean> isTerminated = SettableFuture.create();
 
     @Override
-    public void transportShutdown(Status shutdownStatus) {
+    public void transportShutdown(Status shutdownStatus, DisconnectError disconnectError) {
       if (!this.shutdownStatus.set(shutdownStatus)) {
         throw new IllegalStateException("transportShutdown() already called");
       }

binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java

@@ -56,6 +56,7 @@
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.ManagedClientTransport;
 import io.grpc.internal.ObjectPool;
+import io.grpc.internal.SimpleDisconnectError;
 import io.grpc.internal.StatsTraceContext;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
@@ -305,7 +306,7 @@ public synchronized void shutdownNow(Status reason) {
   @Override
   @GuardedBy("this")
   void notifyShutdown(Status status) {
-    clientTransportListener.transportShutdown(status);
+    clientTransportListener.transportShutdown(status, SimpleDisconnectError.UNKNOWN);
   }
 
   @Override

binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java

@@ -26,6 +26,7 @@
 import static io.grpc.binder.internal.BinderTransport.WIRE_FORMAT_VERSION;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.Assume.assumeTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -58,6 +59,7 @@
 import io.grpc.internal.ClientTransport;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
 import io.grpc.internal.ConnectionClientTransport;
+import io.grpc.internal.DisconnectError;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.InternalServer;
 import io.grpc.internal.ManagedClientTransport;
@@ -357,7 +359,7 @@ public void clientIgnoresTransactionFromNonServerUids() throws Exception {
     sendShutdownTransportTransactionAsUid(client, serverUid);
 
     verify(mockClientTransportListener, timeout(TIMEOUT_MS))
-        .transportShutdown(statusCaptor.capture());
+        .transportShutdown(statusCaptor.capture(), any(DisconnectError.class));
     assertThat(statusCaptor.getValue().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
     assertThat(statusCaptor.getValue().getDescription()).contains("shutdown");
   }
@@ -386,7 +388,7 @@ public void clientReportsAuthzErrorToServer() throws Exception {
             .build();
     runIfNotNull(client.start(mockClientTransportListener));
     verify(mockClientTransportListener, timeout(TIMEOUT_MS))
-        .transportShutdown(statusCaptor.capture());
+        .transportShutdown(statusCaptor.capture(), any(DisconnectError.class));
     assertThat(statusCaptor.getValue().getCode()).isEqualTo(Status.Code.PERMISSION_DENIED);
 
     // Client doesn't tell the server in this case by design -- we don't even want to start it!

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

@@ -201,8 +201,8 @@ public ListenableFuture<SocketStats> getStats() {
   }
 
   /**
-   * Prevents creating any new streams.  Buffered streams are not failed and may still proceed
-   * when {@link #reprocess} is called.  The delayed transport will be terminated when there is no
+   * Prevents creating any new streams. Buffered streams are not failed and may still proceed
+   * when {@link #reprocess} is called. The delayed transport will be terminated when there is no
    * more buffered streams.
    */
   @Override
@@ -215,7 +215,7 @@ public final void shutdown(final Status status) {
       syncContext.executeLater(new Runnable() {
           @Override
           public void run() {
-            listener.transportShutdown(status);
+            listener.transportShutdown(status, SimpleDisconnectError.SUBCHANNEL_SHUTDOWN);
           }
         });
       if (!hasPendingStreams() && reportTransportTerminated != null) {

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed 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 io.grpc.internal;
+
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * Represents the reason for a subchannel disconnection.
+ * Implementations are either the SimpleDisconnectError enum or the GoAwayDisconnectError class for
+ * dynamic ones.
+ */
+@Immutable
+public interface DisconnectError {
+  /**
+   * Returns the string representation suitable for use as an error tag.
+   *
+   * @return The formatted error tag string.
+   */
+  String toErrorString();
+}

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed 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 io.grpc.internal;
+
+
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * Represents a dynamic disconnection due to an HTTP/2 GOAWAY frame.
+ * This class is immutable and holds the specific error code from the frame.
+ */
+@Immutable
+public final class GoAwayDisconnectError implements DisconnectError {
+  private static final String ERROR_TAG = "GOAWAY";
+  private final GrpcUtil.Http2Error errorCode;
+
+  /**
+   * Creates a GoAway reason.
+   *
+   * @param errorCode The specific, non-null HTTP/2 error code (e.g., "NO_ERROR").
+   */
+  public GoAwayDisconnectError(GrpcUtil.Http2Error errorCode) {
+    if (errorCode == null) {
+      throw new NullPointerException("Http2Error cannot be null for GOAWAY");
+    }
+    this.errorCode = errorCode;
+  }
+
+  @Override
+  public String toErrorString() {
+    return ERROR_TAG + " " + errorCode.name();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    GoAwayDisconnectError goAwayDisconnectError = (GoAwayDisconnectError) o;
+    return errorCode == goAwayDisconnectError.errorCode;
+  }
+
+  @Override
+  public int hashCode() {
+    return errorCode.hashCode();
+  }
+}

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

@@ -326,7 +326,7 @@ public void run() {
   }
 
   /**
-   * Immediately attempt to reconnect if the current state is TRANSIENT_FAILURE. Otherwise this
+   * Immediately attempt to reconnect if the current state is TRANSIENT_FAILURE. Otherwise, this
    * method has no effect.
    */
   void resetConnectBackoff() {
@@ -620,7 +620,7 @@ public void transportInUse(boolean inUse) {
     }
 
     @Override
-    public void transportShutdown(final Status s) {
+    public void transportShutdown(final Status s, final DisconnectError disconnectError) {
       channelLogger.log(
           ChannelLogLevel.INFO, "{0} SHUTDOWN with {1}", transport.getLogId(), printShortStatus(s));
       shutdownInitiated = true;
@@ -639,8 +639,7 @@ public void run() {
                     NameResolver.ATTR_BACKEND_SERVICE),
                 /* locality= */ getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                     EquivalentAddressGroup.ATTR_LOCALITY_NAME),
-                /* disconnectError= */ SubchannelMetrics.DisconnectError.UNKNOWN
-                    .getErrorString(null),
+                /* disconnectError= */ disconnectError.toErrorString(),
                 /* securityLevel= */ extractSecurityLevel(addressIndex.getCurrentEagAttributes()
                     .get(GrpcAttributes.ATTR_SECURITY_LEVEL)));
           } else if (pendingTransport == transport) {

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

@@ -27,6 +27,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
 
 /**
  * Manages keepalive pings.
@@ -262,9 +263,25 @@ public interface KeepAlivePinger {
    * Default client side {@link KeepAlivePinger}.
    */
   public static final class ClientKeepAlivePinger implements KeepAlivePinger {
-    private final ConnectionClientTransport transport;
 
-    public ClientKeepAlivePinger(ConnectionClientTransport transport) {
+
+    /**
+     * A {@link ClientTransport} that has life-cycle management.
+     *
+     */
+    @ThreadSafe
+    public interface TransportWithDisconnectReason extends ClientTransport {
+
+      /**
+       * Initiates a forceful shutdown in which preexisting and new calls are closed. Existing calls
+       * should be closed with the provided {@code reason} and {@code disconnectError}.
+       */
+      void shutdownNow(Status reason, DisconnectError disconnectError);
+    }
+
+    private final TransportWithDisconnectReason transport;
+
+    public ClientKeepAlivePinger(TransportWithDisconnectReason transport) {
       this.transport = transport;
     }
 
@@ -277,15 +294,17 @@ public void onSuccess(long roundTripTimeNanos) {}
         @Override
         public void onFailure(Status cause) {
           transport.shutdownNow(Status.UNAVAILABLE.withDescription(
-              "Keepalive failed. The connection is likely gone"));
+                  "Keepalive failed. The connection is likely gone"),
+              SimpleDisconnectError.CONNECTION_TIMED_OUT);
         }
       }, MoreExecutors.directExecutor());
     }
 
     @Override
     public void onPingTimeout() {
       transport.shutdownNow(Status.UNAVAILABLE.withDescription(
-          "Keepalive failed. The connection is likely gone"));
+              "Keepalive failed. The connection is likely gone"),
+          SimpleDisconnectError.CONNECTION_TIMED_OUT);
     }
   }
 }

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

@@ -2056,7 +2056,7 @@ public String toString() {
    */
   private final class DelayedTransportListener implements ManagedClientTransport.Listener {
     @Override
-    public void transportShutdown(Status s) {
+    public void transportShutdown(Status s, DisconnectError e) {
       checkState(shutdown.get(), "Channel must have been shut down");
     }
 

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

@@ -77,8 +77,9 @@ interface Listener {
      * <p>This is called exactly once, and must be called prior to {@link #transportTerminated}.
      *
      * @param s the reason for the shutdown.
+     * @param e the disconnect error.
      */
-    void transportShutdown(Status s);
+    void transportShutdown(Status s, DisconnectError e);
 
     /**
      * The transport completed shutting down. All resources have been released. All streams have