[ISSUE #11474] feat(proxy): expose call metrics for the admin gRPC interface

proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java

@@ -42,6 +42,7 @@
 import org.apache.rocketmq.proxy.grpc.admin.ProxyAdminAuthInterceptor;
 import org.apache.rocketmq.proxy.grpc.admin.ProxyAdminForwarder;
 import org.apache.rocketmq.proxy.grpc.admin.ProxyAdminGrpcService;
+import org.apache.rocketmq.proxy.grpc.admin.ProxyAdminMetricsInterceptor;
 import org.apache.rocketmq.proxy.grpc.v2.DefaultGrpcMessagingActivity;
 import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingApplication;
 import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
@@ -127,9 +128,11 @@ public static void main(String[] args) {
                 GrpcServer adminGrpcServer = GrpcServerBuilder.newBuilder(executor, adminPort, tlsCertificateManager)
                     .addService(adminService)
                     // authentication has to see the channel id HeaderInterceptor derives from the
-                    // transport, so it is ordered after the standard pipeline rather than appended
+                    // transport, so it is ordered after the standard pipeline rather than appended;
+                    // metrics is registered last so it runs before auth and counts rejected calls
                     .configInterceptor(new ProxyAdminAuthInterceptor(
-                        ConfigurationManager.getAuthConfig(), messagingProcessor))
+                        ConfigurationManager.getAuthConfig(), messagingProcessor),
+                        new ProxyAdminMetricsInterceptor())
                     .shutdownTime(ConfigurationManager.getProxyConfig().getGrpcShutdownTimeSeconds(), TimeUnit.SECONDS)
                     .build();
                 PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(adminGrpcServer);

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptor.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.rocketmq.proxy.grpc.admin;
+
+import io.grpc.ForwardingServerCall;
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.rocketmq.proxy.metrics.ProxyMetricsManager;
+
+/**
+ * Records every admin RPC served by the admin gRPC server: one counter increment and one latency
+ * sample per call, labeled with the method and the terminal status code, so call rate, RT and
+ * error rate of the control-plane surface are visible in the same OpenTelemetry pipeline as the
+ * rest of the proxy.
+ *
+ * <p>Registered after {@link ProxyAdminAuthInterceptor} in {@code configInterceptor(...)}; gRPC runs
+ * interceptors in reverse registration order, so this one wraps the call from the outside and also
+ * records calls that authentication or authorization rejects.
+ *
+ * <p>{@code GlobalExceptionInterceptor} sits outside this interceptor and closes the call it holds
+ * directly, bypassing the {@code close} wrapper below, so a service that throws from a listener
+ * callback is recorded here before the exception is rethrown to that interceptor.
+ */
+public class ProxyAdminMetricsInterceptor implements ServerInterceptor {
+
+    @Override
+    public <R, W> ServerCall.Listener<R> interceptCall(ServerCall<R, W> call, Metadata headers,
+        ServerCallHandler<R, W> next) {
+        long startNanos = System.nanoTime();
+        String method = call.getMethodDescriptor().getBareMethodName() != null
+            ? call.getMethodDescriptor().getBareMethodName()
+            : call.getMethodDescriptor().getFullMethodName();
+        // A client cancel reaches the listener while a handler may still close the call later;
+        // the first terminal event wins so a call is never counted twice.
+        AtomicBoolean recorded = new AtomicBoolean(false);
+
+        ServerCall<R, W> wrapped = new ForwardingServerCall.SimpleForwardingServerCall<R, W>(call) {
+            @Override
+            public void close(Status status, Metadata trailers) {
+                record(recorded, method, status.getCode().name(), startNanos);
+                super.close(status, trailers);
+            }
+        };
+
+        ServerCall.Listener<R> listener = next.startCall(wrapped, headers);
+        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<R>(listener) {
+            @Override
+            public void onMessage(R message) {
+                try {
+                    super.onMessage(message);
+                } catch (RuntimeException | Error e) {
+                    record(recorded, method, statusCodeOf(e), startNanos);
+                    throw e;
+                }
+            }
+
+            @Override
+            public void onHalfClose() {
+                try {
+                    super.onHalfClose();
+                } catch (RuntimeException | Error e) {
+                    record(recorded, method, statusCodeOf(e), startNanos);
+                    throw e;
+                }
+            }
+
+            @Override
+            public void onCancel() {
+                record(recorded, method, Status.CANCELLED.getCode().name(), startNanos);
+                super.onCancel();
+            }
+        };
+    }
+
+    private static String statusCodeOf(Throwable t) {
+        return t instanceof StatusRuntimeException
+            ? ((StatusRuntimeException) t).getStatus().getCode().name()
+            : Status.INTERNAL.getCode().name();
+    }
+
+    private static void record(AtomicBoolean recorded, String method, String status, long startNanos) {
+        if (recorded.compareAndSet(false, true)) {
+            ProxyMetricsManager.recordAdminCall(method, status,
+                TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos));
+        }
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java

@@ -19,6 +19,10 @@
 public class ProxyMetricsConstant {
     public static final String GAUGE_PROXY_UP = "rocketmq_proxy_up";
 
+    public static final String COUNTER_PROXY_ADMIN_CALLS_TOTAL = "rocketmq_proxy_admin_calls_total";
+    public static final String HISTOGRAM_PROXY_ADMIN_RT = "rocketmq_proxy_admin_rt";
+
     public static final String LABEL_PROXY_MODE = "proxy_mode";
+    public static final String LABEL_METHOD = "method";
     public static final String NODE_TYPE_PROXY = "proxy";
 }

proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java

@@ -19,6 +19,8 @@
 import com.google.common.base.Splitter;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.ObservableLongGauge;
 import io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingMetricExporter;
@@ -36,12 +38,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.metrics.MetricsExporterType;
+import org.apache.rocketmq.common.metrics.NopLongCounter;
+import org.apache.rocketmq.common.metrics.NopLongHistogram;
 import org.apache.rocketmq.common.utils.StartAndShutdown;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -51,18 +56,24 @@
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.AGGREGATION_DELTA;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_AGGREGATION;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CLUSTER_NAME;
+import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_INVOCATION_STATUS;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_ID;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_TYPE;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.OPEN_TELEMETRY_METER_NAME;
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.COUNTER_PROXY_ADMIN_CALLS_TOTAL;
 import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.GAUGE_PROXY_UP;
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.HISTOGRAM_PROXY_ADMIN_RT;
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.LABEL_METHOD;
 import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.LABEL_PROXY_MODE;
 import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.NODE_TYPE_PROXY;
 
 public class ProxyMetricsManager implements StartAndShutdown {
     private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
 
     private static ProxyConfig proxyConfig;
-    private final static Map<String, String> LABEL_MAP = new HashMap<>();
+    // Concurrent: admin RPC recording reads this on request threads while start() populates it —
+    // in cluster mode the gRPC servers are appended to the start list before the metrics manager.
+    private final static Map<String, String> LABEL_MAP = new ConcurrentHashMap<>();
     public static Supplier<AttributesBuilder> attributesBuilderSupplier;
 
     private OtlpGrpcMetricExporter metricExporter;
@@ -72,6 +83,14 @@ public class ProxyMetricsManager implements StartAndShutdown {
 
     public static ObservableLongGauge proxyUp = null;
 
+    /**
+     * Admin RPC instrumentation. Both default to no-ops so that calls served before the metrics
+     * manager starts (or with metrics disabled entirely) are simply not recorded instead of
+     * tripping a null check; {@link #initMetrics} swaps in the real instruments when enabled.
+     */
+    public static LongCounter proxyAdminCallTotal = new NopLongCounter();
+    public static LongHistogram proxyAdminRt = new NopLongHistogram();
+
     public static void initLocalMode(BrokerMetricsManager brokerMetricsManager, ProxyConfig proxyConfig) {
         if (proxyConfig.getMetricsExporterType() == MetricsExporterType.DISABLE) {
             return;
@@ -108,6 +127,30 @@ private static void initMetrics(Meter meter, Supplier<AttributesBuilder> attribu
             .setDescription("proxy status")
             .ofLongs()
             .buildWithCallback(measurement -> measurement.record(1, newAttributesBuilder().build()));
+
+        proxyAdminCallTotal = meter.counterBuilder(COUNTER_PROXY_ADMIN_CALLS_TOTAL)
+            .setDescription("proxy admin RPC count by method and status")
+            .build();
+        proxyAdminRt = meter.histogramBuilder(HISTOGRAM_PROXY_ADMIN_RT)
+            .setDescription("proxy admin RPC latency by method and status")
+            .ofLongs()
+            .setUnit("milliseconds")
+            .build();
+    }
+
+    /**
+     * Records one admin RPC: a counter increment and a latency sample, both labeled with the bare
+     * gRPC method name, the terminal status code and the node attributes from
+     * {@link #newAttributesBuilder()}. Safe to call at any time — before initialization or with
+     * metrics disabled it lands in the no-op instruments.
+     */
+    public static void recordAdminCall(String method, String invocationStatus, long durationMs) {
+        Attributes attributes = newAttributesBuilder()
+            .put(LABEL_METHOD, method)
+            .put(LABEL_INVOCATION_STATUS, invocationStatus)
+            .build();
+        proxyAdminCallTotal.add(1, attributes);
+        proxyAdminRt.record(durationMs, attributes);
     }
 
     public ProxyMetricsManager() {

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerInterceptorOrderTest.java

@@ -79,6 +79,32 @@ public void postHeaderInterceptorsAreOrderedAfterHeaderInterceptorTest() {
             authIndex < headerIndex);
     }
 
+    /**
+     * The admin server registers two post-header interceptors, auth first and metrics second
+     * ({@code ProxyStartup}). Reversed execution therefore makes metrics the outer wrapper, which is
+     * what lets it observe the status auth closes a rejected call with. Pin the relative order so a
+     * future re-registration cannot silently stop counting rejected admin calls.
+     */
+    @Test
+    public void metricsInterceptorIsRegisteredAfterAuthInterceptorTest() {
+        ServerInterceptor authInterceptor = mock(ServerInterceptor.class);
+        ServerInterceptor metricsInterceptor = mock(ServerInterceptor.class);
+        @SuppressWarnings("unchecked")
+        ServerBuilder<?> serverBuilder = mock(ServerBuilder.class);
+
+        GrpcServerBuilder.configureInterceptors(serverBuilder, authInterceptor, metricsInterceptor);
+
+        ArgumentCaptor<ServerInterceptor> captor = ArgumentCaptor.forClass(ServerInterceptor.class);
+        verify(serverBuilder, atLeastOnce()).intercept(captor.capture());
+        List<ServerInterceptor> registrationOrder = captor.getAllValues();
+
+        assertSame("auth is registered first", authInterceptor, registrationOrder.get(0));
+        assertSame("metrics is registered second", metricsInterceptor, registrationOrder.get(1));
+        // registered later means executed earlier: metrics wraps the call, auth closes through it
+        assertTrue("metrics must run before auth so rejected closes are recorded",
+            registrationOrder.indexOf(authInterceptor) < registrationOrder.indexOf(metricsInterceptor));
+    }
+
     /**
      * The normalization the ordering above depends on: whatever channel id the client puts on the
      * wire is discarded in favour of the one derived from the transport.