core,opentelemetry: Fix early-close metric labels

Resolve generated-method classification before serverCallStarted() so close metrics do not fall back
 to "other" when streamClosed() happens first.

Keep fallback registry lookup on the existing async
 path and avoid tracer-side HandlerRegistry access to match the maintainer constraints from issue #12117.

Add regressions for primary generated, primary non-generated, and fallback-generated server paths,
 plus the tracer-level early-resolution contract.

Author: becomeStar

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

@@ -99,7 +99,7 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
   private final ObjectPool<? extends Executor> executorPool;
   /** Executor for application processing. Safe to read after {@link #start()}. */
   private Executor executor;
-  private final HandlerRegistry registry;
+  private final InternalHandlerRegistry registry;
   private final HandlerRegistry fallbackRegistry;
   private final List<ServerTransportFilter> transportFilters;
   // This is iterated on a per-call basis.  Use an array instead of a Collection to avoid iterator
@@ -498,8 +498,12 @@ private void streamCreatedInternal(
 
       final StatsTraceContext statsTraceCtx = Preconditions.checkNotNull(
           stream.statsTraceContext(), "statsTraceCtx not present from stream");
+      final ServerMethodDefinition<?, ?> primaryMethod = registry.lookupMethod(methodName, null);
 
       final Context.CancellableContext context = createContext(headers, statsTraceCtx);
+      if (primaryMethod != null) {
+        statsTraceCtx.serverCallMethodResolved(primaryMethod.getMethodDescriptor());
+      }
 
       final Link link = PerfMark.linkOut();
 
@@ -536,7 +540,7 @@ private void runInternal() {
           ServerMethodDefinition<?, ?> wrapMethod;
           ServerCallParameters<?, ?> callParams;
           try {
-            ServerMethodDefinition<?, ?> method = registry.lookupMethod(methodName);
+            ServerMethodDefinition<?, ?> method = primaryMethod;
             if (method == null) {
               method = fallbackRegistry.lookupMethod(methodName, stream.getAuthority());
             }
@@ -554,7 +558,12 @@ private void runInternal() {
               future.cancel(false);
               return;
             }
-            wrapMethod = wrapMethod(stream, method, statsTraceCtx);
+            statsTraceCtx.serverCallStarted(
+                new ServerCallInfoImpl<>(
+                    method.getMethodDescriptor(), // notify with original method descriptor
+                    stream.getAttributes(),
+                    stream.getAuthority()));
+            wrapMethod = wrapMethod(method);
             callParams = maySwitchExecutor(wrapMethod, stream, headers, context, tag);
             future.set(callParams);
           } catch (Throwable t) {
@@ -653,14 +662,8 @@ private Context.CancellableContext createContext(
     }
 
     /** Never returns {@code null}. */
-    private <ReqT, RespT> ServerMethodDefinition<?,?> wrapMethod(ServerStream stream,
-        ServerMethodDefinition<ReqT, RespT> methodDef, StatsTraceContext statsTraceCtx) {
-      // TODO(ejona86): should we update fullMethodName to have the canonical path of the method?
-      statsTraceCtx.serverCallStarted(
-          new ServerCallInfoImpl<>(
-              methodDef.getMethodDescriptor(), // notify with original method descriptor
-              stream.getAttributes(),
-              stream.getAuthority()));
+    private <ReqT, RespT> ServerMethodDefinition<?,?> wrapMethod(
+        ServerMethodDefinition<ReqT, RespT> methodDef) {
       ServerCallHandler<ReqT, RespT> handler = methodDef.getServerCallHandler();
       for (ServerInterceptor interceptor : interceptors) {
         handler = InternalServerInterceptors.interceptCallHandlerCreate(interceptor, handler);

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

@@ -23,6 +23,7 @@
 import io.grpc.ClientStreamTracer;
 import io.grpc.Context;
 import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
 import io.grpc.ServerStreamTracer;
 import io.grpc.ServerStreamTracer.ServerCallInfo;
 import io.grpc.Status;
@@ -38,6 +39,14 @@
  */
 @ThreadSafe
 public final class StatsTraceContext {
+  /**
+   * Internal hook for server tracers that can use the resolved method descriptor before
+   * {@link ServerStreamTracer#serverCallStarted(ServerCallInfo)} runs.
+   */
+  public interface ServerCallMethodListener {
+    void serverCallMethodResolved(MethodDescriptor<?, ?> method);
+  }
+
   public static final StatsTraceContext NOOP = new StatsTraceContext(new StreamTracer[0]);
 
   private final StreamTracer[] tracers;
@@ -144,6 +153,20 @@ public void serverCallStarted(ServerCallInfo<?, ?> callInfo) {
     }
   }
 
+  /**
+   * Notifies server tracers that a primary-registry method descriptor was resolved before
+   * {@link ServerStreamTracer#serverCallStarted(ServerCallInfo)}.
+   *
+   * <p>Called from {@link io.grpc.internal.ServerImpl}.
+   */
+  public void serverCallMethodResolved(MethodDescriptor<?, ?> method) {
+    for (StreamTracer tracer : tracers) {
+      if (tracer instanceof ServerCallMethodListener) {
+        ((ServerCallMethodListener) tracer).serverCallMethodResolved(method);
+      }
+    }
+  }
+
   /**
    * See {@link StreamTracer#streamClosed}. This may be called multiple times, and only the first
    * value will be taken.

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

@@ -129,6 +129,10 @@ public class ServerImplTest {
           .setRequestMarshaller(STRING_MARSHALLER)
           .setResponseMarshaller(INTEGER_MARSHALLER)
           .build();
+  private static final MethodDescriptor<String, Integer> GENERATED_METHOD =
+      METHOD.toBuilder()
+          .setSampledToLocalTracing(true)
+          .build();
   private static final Context.Key<String> SERVER_ONLY = Context.key("serverOnly");
   private static final Context.Key<String> SERVER_TRACER_ADDED_KEY = Context.key("tracer-added");
   private static final Context.CancellableContext SERVER_CONTEXT =
@@ -142,6 +146,60 @@ public boolean shouldAccept(Runnable runnable) {
       };
   private static final String AUTHORITY = "some_authority";
 
+  private static final class MethodNameCapturingTracer extends ServerStreamTracer
+      implements StatsTraceContext.ServerCallMethodListener {
+    @Nullable private ServerCallInfo<?, ?> serverCallInfo;
+    @Nullable private String recordedMethodName;
+    @Nullable private String resolvedMethodName;
+    private boolean streamClosed;
+
+    @Override
+    public synchronized void serverCallMethodResolved(MethodDescriptor<?, ?> method) {
+      resolvedMethodName =
+          recordMethodName(method.isSampledToLocalTracing(), method.getFullMethodName());
+    }
+
+    @Override
+    public synchronized void streamClosed(Status status) {
+      streamClosed = true;
+      if (serverCallInfo != null) {
+        recordedMethodName =
+            recordMethodName(
+                serverCallInfo.getMethodDescriptor().isSampledToLocalTracing(),
+                serverCallInfo.getMethodDescriptor().getFullMethodName());
+      } else if (resolvedMethodName != null) {
+        recordedMethodName = resolvedMethodName;
+      } else {
+        recordedMethodName = "other";
+      }
+    }
+
+    @Override
+    public synchronized void serverCallStarted(ServerCallInfo<?, ?> callInfo) {
+      serverCallInfo = callInfo;
+      if (streamClosed) {
+        recordedMethodName =
+            recordMethodName(
+                callInfo.getMethodDescriptor().isSampledToLocalTracing(),
+                callInfo.getMethodDescriptor().getFullMethodName());
+      }
+    }
+
+    @Nullable
+    synchronized ServerCallInfo<?, ?> getServerCallInfo() {
+      return serverCallInfo;
+    }
+
+    @Nullable
+    synchronized String getRecordedMethodName() {
+      return recordedMethodName;
+    }
+
+    private static String recordMethodName(boolean generatedMethod, String fullMethodName) {
+      return generatedMethod ? fullMethodName : "other";
+    }
+  }
+
   @Rule public final MockitoRule mocks = MockitoJUnit.rule();
 
   @BeforeClass
@@ -462,6 +520,172 @@ public void methodNotFound() throws Exception {
     assertEquals(Status.Code.UNIMPLEMENTED, statusCaptor.getValue().getCode());
   }
 
+  @Test
+  public void primaryRegistryGeneratedMethod_streamClosedBeforeStart_preservesMethodName()
+      throws Exception {
+    MethodNameCapturingTracer methodNameTracer = new MethodNameCapturingTracer();
+    streamTracerFactories =
+        Collections.singletonList(
+            new ServerStreamTracer.Factory() {
+              @Override
+              public ServerStreamTracer newServerStreamTracer(
+                  String fullMethodName, Metadata headers) {
+                return methodNameTracer;
+              }
+            });
+    builder.addService(
+        ServerServiceDefinition.builder(new ServiceDescriptor("Waiter", GENERATED_METHOD))
+            .addMethod(
+                GENERATED_METHOD,
+                new ServerCallHandler<String, Integer>() {
+                  @Override
+                  public ServerCall.Listener<String> startCall(
+                      ServerCall<String, Integer> call, Metadata headers) {
+                    return callListener;
+                  }
+                })
+            .build());
+
+    createAndStartServer();
+    ServerTransportListener transportListener
+        = transportServer.registerNewServerTransport(new SimpleServerTransport());
+    transportListener.transportReady(Attributes.EMPTY);
+    Metadata requestHeaders = new Metadata();
+    StatsTraceContext statsTraceCtx =
+        StatsTraceContext.newServerContext(
+            streamTracerFactories, GENERATED_METHOD.getFullMethodName(), requestHeaders);
+    when(stream.getAttributes()).thenReturn(Attributes.EMPTY);
+    when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
+
+    transportListener.streamCreated(stream, GENERATED_METHOD.getFullMethodName(), requestHeaders);
+    verify(stream).setListener(isA(ServerStreamListener.class));
+    verify(stream, atLeast(1)).statsTraceContext();
+
+    statsTraceCtx.streamClosed(Status.CANCELLED);
+    assertNull(methodNameTracer.getServerCallInfo());
+    assertEquals(
+        GENERATED_METHOD.getFullMethodName(),
+        methodNameTracer.getRecordedMethodName());
+
+    assertEquals(1, executor.runDueTasks());
+
+    assertNotNull(methodNameTracer.getServerCallInfo());
+    assertSame(GENERATED_METHOD, methodNameTracer.getServerCallInfo().getMethodDescriptor());
+    assertEquals(
+        GENERATED_METHOD.getFullMethodName(),
+        methodNameTracer.getRecordedMethodName());
+    verify(fallbackRegistry, never()).lookupMethod(anyString(), any());
+  }
+
+  @Test
+  public void primaryRegistryNonGeneratedMethod_streamClosedBeforeStart_recordsOther()
+      throws Exception {
+    MethodNameCapturingTracer methodNameTracer = new MethodNameCapturingTracer();
+    streamTracerFactories =
+        Collections.singletonList(
+            new ServerStreamTracer.Factory() {
+              @Override
+              public ServerStreamTracer newServerStreamTracer(
+                  String fullMethodName, Metadata headers) {
+                return methodNameTracer;
+              }
+            });
+    builder.addService(
+        ServerServiceDefinition.builder(new ServiceDescriptor("Waiter", METHOD))
+            .addMethod(
+                METHOD,
+                new ServerCallHandler<String, Integer>() {
+                  @Override
+                  public ServerCall.Listener<String> startCall(
+                      ServerCall<String, Integer> call, Metadata headers) {
+                    return callListener;
+                  }
+                })
+            .build());
+
+    createAndStartServer();
+    ServerTransportListener transportListener
+        = transportServer.registerNewServerTransport(new SimpleServerTransport());
+    transportListener.transportReady(Attributes.EMPTY);
+    Metadata requestHeaders = new Metadata();
+    StatsTraceContext statsTraceCtx =
+        StatsTraceContext.newServerContext(
+            streamTracerFactories, METHOD.getFullMethodName(), requestHeaders);
+    when(stream.getAttributes()).thenReturn(Attributes.EMPTY);
+    when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
+
+    transportListener.streamCreated(stream, METHOD.getFullMethodName(), requestHeaders);
+    verify(stream).setListener(isA(ServerStreamListener.class));
+    verify(stream, atLeast(1)).statsTraceContext();
+
+    statsTraceCtx.streamClosed(Status.CANCELLED);
+    assertNull(methodNameTracer.getServerCallInfo());
+    assertEquals("other", methodNameTracer.getRecordedMethodName());
+
+    assertEquals(1, executor.runDueTasks());
+
+    assertNotNull(methodNameTracer.getServerCallInfo());
+    assertSame(METHOD, methodNameTracer.getServerCallInfo().getMethodDescriptor());
+    assertEquals("other", methodNameTracer.getRecordedMethodName());
+    verify(fallbackRegistry, never()).lookupMethod(anyString(), any());
+  }
+
+  @Test
+  public void fallbackRegistryGeneratedMethod_streamClosedBeforeStart_resolvesOnAsyncLookup()
+      throws Exception {
+    MethodNameCapturingTracer methodNameTracer = new MethodNameCapturingTracer();
+    streamTracerFactories =
+        Collections.singletonList(
+            new ServerStreamTracer.Factory() {
+              @Override
+              public ServerStreamTracer newServerStreamTracer(
+                  String fullMethodName, Metadata headers) {
+                return methodNameTracer;
+              }
+            });
+    mutableFallbackRegistry.addService(
+        ServerServiceDefinition.builder(new ServiceDescriptor("Waiter", GENERATED_METHOD))
+            .addMethod(
+                GENERATED_METHOD,
+                new ServerCallHandler<String, Integer>() {
+                  @Override
+                  public ServerCall.Listener<String> startCall(
+                      ServerCall<String, Integer> call, Metadata headers) {
+                    return callListener;
+                  }
+                })
+            .build());
+
+    createAndStartServer();
+    ServerTransportListener transportListener
+        = transportServer.registerNewServerTransport(new SimpleServerTransport());
+    transportListener.transportReady(Attributes.EMPTY);
+    Metadata requestHeaders = new Metadata();
+    StatsTraceContext statsTraceCtx =
+        StatsTraceContext.newServerContext(
+            streamTracerFactories, GENERATED_METHOD.getFullMethodName(), requestHeaders);
+    when(stream.getAttributes()).thenReturn(Attributes.EMPTY);
+    when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
+
+    transportListener.streamCreated(stream, GENERATED_METHOD.getFullMethodName(), requestHeaders);
+    verify(stream).setListener(isA(ServerStreamListener.class));
+    verify(stream, atLeast(1)).statsTraceContext();
+
+    statsTraceCtx.streamClosed(Status.CANCELLED);
+    assertNull(methodNameTracer.getServerCallInfo());
+    assertEquals("other", methodNameTracer.getRecordedMethodName());
+    verify(fallbackRegistry, never()).lookupMethod(anyString(), any());
+
+    assertEquals(1, executor.runDueTasks());
+
+    assertNotNull(methodNameTracer.getServerCallInfo());
+    assertSame(GENERATED_METHOD, methodNameTracer.getServerCallInfo().getMethodDescriptor());
+    assertEquals(
+        GENERATED_METHOD.getFullMethodName(),
+        methodNameTracer.getRecordedMethodName());
+    verify(fallbackRegistry).lookupMethod(GENERATED_METHOD.getFullMethodName(), AUTHORITY);
+  }
+
 
   @Test
   public void executorSupplierSameExecutorBasic() throws Exception {