From 272b558f756c6cc1808fcfe0053c9405e5eb4622 Mon Sep 17 00:00:00 2001 From: Felix-Gong Date: Tue, 20 Oct 2026 15:58:17 +0000 Subject: [PATCH] [ISSUE #11474] feat(proxy): expose call metrics for the admin gRPC interface The proxy exports an OpenTelemetry pipeline in both modes but registers only the rocketmq_proxy_up gauge: call rate, latency and error rate of the admin gRPC server are invisible, so a healthy-looking up gauge says nothing about auth rejections or failing admin calls. Add ProxyAdminMetricsInterceptor to the admin server and two instruments registered through the existing initMetrics path: rocketmq_proxy_admin_calls_total and rocketmq_proxy_admin_rt (milliseconds), labeled with the bare method name and the terminal invocation_status on top of the node/cluster/proxy_mode attributes. The interceptor wraps close(Status) and records exactly once per call (the cancel/close race is settled by an AtomicBoolean), and is registered after the auth interceptor so gRPC reverse-order execution makes it the outer wrapper that also counts rejected calls. Listener callbacks that throw are recorded before the exception is rethrown, because GlobalExceptionInterceptor closes the call it holds directly and would otherwise bypass the close wrapper. Instruments default to NopLongCounter/NopLongHistogram, so disabled metrics or a call served before manager start is a free no-op. LABEL_MAP becomes a ConcurrentHashMap: recording now reads it on request threads while start() populates it. Tests: ProxyAdminMetricsInterceptorTest 7/7 (OK and PERMISSION_DENIED recording, cancel/close dedup in both orders, delegate close propagation, builtin method full name, thrown exceptions), a test MetricReader asserts the exported names, labels and values, GrpcServerInterceptorOrderTest pins the auth/metrics relative order. Full proxy module: 465 green with checkstyle, spotbugs and misspell. --- .../apache/rocketmq/proxy/ProxyStartup.java | 7 +- .../admin/ProxyAdminMetricsInterceptor.java | 108 ++++++++++ .../proxy/metrics/ProxyMetricsConstant.java | 4 + .../proxy/metrics/ProxyMetricsManager.java | 45 +++- .../grpc/GrpcServerInterceptorOrderTest.java | 26 +++ .../ProxyAdminMetricsInterceptorTest.java | 170 +++++++++++++++ .../metrics/ProxyAdminCallMetricsTest.java | 204 ++++++++++++++++++ 7 files changed, 561 insertions(+), 3 deletions(-) create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptor.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptorTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/metrics/ProxyAdminCallMetricsTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java index 5c56f0fe55c..f7c5e0f82af 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java +++ b/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); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptor.java new file mode 100644 index 00000000000..1b90caa0018 --- /dev/null +++ b/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. + * + *

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. + * + *

{@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 ServerCall.Listener interceptCall(ServerCall call, Metadata headers, + ServerCallHandler 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 wrapped = new ForwardingServerCall.SimpleForwardingServerCall(call) { + @Override + public void close(Status status, Metadata trailers) { + record(recorded, method, status.getCode().name(), startNanos); + super.close(status, trailers); + } + }; + + ServerCall.Listener listener = next.startCall(wrapped, headers); + return new ForwardingServerCallListener.SimpleForwardingServerCallListener(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)); + } + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java index a7682e6cf93..db6c2f15e55 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java +++ b/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"; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java index 81db576e3d2..7827fd8e87d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java +++ b/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,10 +56,14 @@ 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; @@ -62,7 +71,9 @@ public class ProxyMetricsManager implements StartAndShutdown { private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private static ProxyConfig proxyConfig; - private final static Map 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 LABEL_MAP = new ConcurrentHashMap<>(); public static Supplier 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 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() { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerInterceptorOrderTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerInterceptorOrderTest.java index 2c1044342cb..cf2add71735 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerInterceptorOrderTest.java +++ b/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 captor = ArgumentCaptor.forClass(ServerInterceptor.class); + verify(serverBuilder, atLeastOnce()).intercept(captor.capture()); + List 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. diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptorTest.java new file mode 100644 index 00000000000..99a89526119 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/admin/ProxyAdminMetricsInterceptorTest.java @@ -0,0 +1,170 @@ +/* + * 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.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.apache.rocketmq.proxy.metrics.ProxyMetricsManager; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class ProxyAdminMetricsInterceptorTest { + + private static final String METHOD = "PrintThreadStackTrace"; + + @Mock + private ServerCall call; + @Mock + private ServerCallHandler next; + @Mock + private ServerCall.Listener innerListener; + + private ProxyAdminMetricsInterceptor interceptor; + private ServerCall wrapped; + private ServerCall.Listener outerListener; + + @Before + @SuppressWarnings("unchecked") + public void setUp() { + interceptor = new ProxyAdminMetricsInterceptor(); + MethodDescriptor descriptor = mock(MethodDescriptor.class); + when(descriptor.getBareMethodName()).thenReturn(METHOD); + when(call.getMethodDescriptor()).thenReturn(descriptor); + when(next.startCall(any(), any())).thenReturn(innerListener); + + outerListener = interceptor.interceptCall(call, new Metadata(), next); + ArgumentCaptor> captor = ArgumentCaptor.forClass(ServerCall.class); + verify(next).startCall(captor.capture(), any()); + wrapped = captor.getValue(); + } + + @Test + public void recordsTerminalStatusOnClose() { + Metadata trailers = new Metadata(); + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + wrapped.close(Status.OK, trailers); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall(eq(METHOD), eq("OK"), anyLong())); + } + verify(call).close(Status.OK, trailers); + } + + @Test + public void recordsRejectedStatus() { + Status denied = Status.PERMISSION_DENIED.withDescription("no acl mapping"); + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + wrapped.close(denied, new Metadata()); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("PERMISSION_DENIED"), anyLong())); + } + verify(call).close(eq(denied), any(Metadata.class)); + } + + @Test + public void cancelRecordedOnceEvenIfHandlerClosesLater() { + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + outerListener.onCancel(); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("CANCELLED"), anyLong()), times(1)); + + wrapped.close(Status.CANCELLED, new Metadata()); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + anyString(), anyString(), anyLong()), times(1)); + } + } + + @Test + public void closeAfterNormalResponseNotCountedByCancel() { + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + wrapped.close(Status.OK, new Metadata()); + outerListener.onCancel(); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("OK"), anyLong()), times(1)); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("CANCELLED"), anyLong()), never()); + } + } + + @Test + public void serviceExceptionRecordedAsInternalAndRethrown() { + doThrow(new IllegalStateException("boom")).when(innerListener).onHalfClose(); + + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + Assert.assertThrows(IllegalStateException.class, () -> outerListener.onHalfClose()); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("INTERNAL"), anyLong()), times(1)); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + anyString(), anyString(), anyLong()), times(1)); + } + } + + @Test + public void serviceStatusExceptionRecordedWithItsOwnCode() { + doThrow(Status.PERMISSION_DENIED.asRuntimeException()).when(innerListener).onHalfClose(); + + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + Assert.assertThrows(StatusRuntimeException.class, () -> outerListener.onHalfClose()); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq(METHOD), eq("PERMISSION_DENIED"), anyLong())); + } + } + + @Test + public void builtinMethodWithoutBareNameRecordsFullMethodName() { + @SuppressWarnings("unchecked") + ServerCall localCall = mock(ServerCall.class); + @SuppressWarnings("unchecked") + ServerCallHandler localNext = mock(ServerCallHandler.class); + MethodDescriptor descriptor = mock(MethodDescriptor.class); + when(descriptor.getBareMethodName()).thenReturn(null); + when(descriptor.getFullMethodName()).thenReturn("grpc.health.v1.Health/Check"); + when(localCall.getMethodDescriptor()).thenReturn(descriptor); + when(localNext.startCall(any(), any())).thenReturn(innerListener); + + new ProxyAdminMetricsInterceptor().interceptCall(localCall, new Metadata(), localNext); + ArgumentCaptor> captor = ArgumentCaptor.forClass(ServerCall.class); + verify(localNext).startCall(captor.capture(), any()); + + try (MockedStatic metrics = mockStatic(ProxyMetricsManager.class)) { + captor.getValue().close(Status.OK, new Metadata()); + metrics.verify(() -> ProxyMetricsManager.recordAdminCall( + eq("grpc.health.v1.Health/Check"), eq("OK"), anyLong())); + } + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/metrics/ProxyAdminCallMetricsTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/metrics/ProxyAdminCallMetricsTest.java new file mode 100644 index 00000000000..aaee3c7eae6 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/metrics/ProxyAdminCallMetricsTest.java @@ -0,0 +1,204 @@ +/* + * 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.metrics; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.LongHistogram; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.CollectionRegistration; +import io.opentelemetry.sdk.metrics.export.MetricReader; +import io.opentelemetry.api.common.AttributesBuilder; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.function.Supplier; +import org.apache.rocketmq.broker.metrics.BrokerMetricsManager; +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.proxy.config.ConfigurationManager; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class ProxyAdminCallMetricsTest { + + @Mock + private BrokerMetricsManager brokerMetricsManager; + + private LongCounter savedCounter; + private LongHistogram savedHistogram; + private MetricsExporterType savedExporterType; + private Supplier savedSupplier; + + @BeforeClass + public static void initConfiguration() throws Exception { + ConfigurationManager.initConfig(); + } + + @Before + public void setUp() { + savedCounter = ProxyMetricsManager.proxyAdminCallTotal; + savedHistogram = ProxyMetricsManager.proxyAdminRt; + savedExporterType = ConfigurationManager.getProxyConfig().getMetricsExporterType(); + savedSupplier = ProxyMetricsManager.attributesBuilderSupplier; + } + + @After + public void tearDown() { + ProxyMetricsManager.proxyAdminCallTotal = savedCounter; + ProxyMetricsManager.proxyAdminRt = savedHistogram; + ProxyMetricsManager.attributesBuilderSupplier = savedSupplier; + ConfigurationManager.getProxyConfig().setMetricsExporterType(savedExporterType); + } + + @Test + public void recordBeforeInitLandsInNoopInstruments() { + ProxyMetricsManager.proxyAdminCallTotal = new NopLongCounter(); + ProxyMetricsManager.proxyAdminRt = new NopLongHistogram(); + + ProxyMetricsManager.recordAdminCall("GetMetadata", "OK", 3); + + assertTrue(ProxyMetricsManager.proxyAdminCallTotal instanceof NopLongCounter); + assertTrue(ProxyMetricsManager.proxyAdminRt instanceof NopLongHistogram); + } + + @Test + public void initLocalModeReplacesNoopInstruments() { + ConfigurationManager.getProxyConfig().setMetricsExporterType(MetricsExporterType.LOG); + Meter meter = OpenTelemetrySdk.builder().build().getMeter("proxy-test"); + when(brokerMetricsManager.getBrokerMeter()).thenReturn(meter); + doReturn(Attributes.builder()).when(brokerMetricsManager).newAttributesBuilder(); + + ProxyMetricsManager.initLocalMode(brokerMetricsManager, ConfigurationManager.getProxyConfig()); + + assertFalse("init must install a real counter", + ProxyMetricsManager.proxyAdminCallTotal instanceof NopLongCounter); + assertFalse("init must install a real histogram", + ProxyMetricsManager.proxyAdminRt instanceof NopLongHistogram); + ProxyMetricsManager.recordAdminCall("GetMetadata", "PERMISSION_DENIED", 7); + } + + @Test + public void disabledMetricsKeepsNoopInstruments() { + ProxyMetricsManager.proxyAdminCallTotal = new NopLongCounter(); + ProxyMetricsManager.proxyAdminRt = new NopLongHistogram(); + ConfigurationManager.getProxyConfig().setMetricsExporterType(MetricsExporterType.DISABLE); + + ProxyMetricsManager.initLocalMode(brokerMetricsManager, ConfigurationManager.getProxyConfig()); + + assertTrue(ProxyMetricsManager.proxyAdminCallTotal instanceof NopLongCounter); + assertTrue(ProxyMetricsManager.proxyAdminRt instanceof NopLongHistogram); + ProxyMetricsManager.recordAdminCall("GetMetadata", "OK", 1); + } + + @Test + public void exportsCounterAndHistogramWithExpectedNamesAndLabels() { + ConfigurationManager.getProxyConfig().setMetricsExporterType(MetricsExporterType.LOG); + CollectingReader reader = new CollectingReader(); + OpenTelemetrySdk sdk = OpenTelemetrySdk.builder() + .setMeterProvider(SdkMeterProvider.builder() + .registerMetricReader(reader) + .build()) + .build(); + Meter meter = sdk.getMeter("proxy-test-e2e"); + when(brokerMetricsManager.getBrokerMeter()).thenReturn(meter); + doReturn(Attributes.builder()).when(brokerMetricsManager).newAttributesBuilder(); + + ProxyMetricsManager.initLocalMode(brokerMetricsManager, ConfigurationManager.getProxyConfig()); + ProxyMetricsManager.recordAdminCall("GetMetadata", "PERMISSION_DENIED", 7); + + AttributeKey methodKey = AttributeKey.stringKey("method"); + AttributeKey statusKey = AttributeKey.stringKey("invocation_status"); + + List all = new ArrayList<>(reader.collect()); + MetricData counter = all.stream() + .filter(m -> "rocketmq_proxy_admin_calls_total".equals(m.getName())) + .findFirst().orElseThrow(() -> new AssertionError("counter not exported: " + names(all))); + long counterValue = counter.getLongSumData().getPoints().iterator().next().getValue(); + Attributes counterAttrs = counter.getLongSumData().getPoints().iterator().next().getAttributes(); + assertEquals(1L, counterValue); + assertEquals("GetMetadata", counterAttrs.get(methodKey)); + assertEquals("PERMISSION_DENIED", counterAttrs.get(statusKey)); + + MetricData histogram = all.stream() + .filter(m -> "rocketmq_proxy_admin_rt".equals(m.getName())) + .findFirst().orElseThrow(() -> new AssertionError("histogram not exported: " + names(all))); + HistogramPointData point = histogram.getHistogramData().getPoints().iterator().next(); + assertEquals(1L, point.getCount()); + assertEquals(7.0, point.getSum(), 0.0001); + assertEquals("GetMetadata", point.getAttributes().get(methodKey)); + assertEquals("PERMISSION_DENIED", point.getAttributes().get(statusKey)); + + sdk.close(); + } + + private static String names(List all) { + List out = new ArrayList<>(); + for (MetricData m : all) { + out.add(m.getName()); + } + return out.toString(); + } + + private static final class CollectingReader implements MetricReader { + private volatile CollectionRegistration registration = CollectionRegistration.noop(); + + @Override + public void register(CollectionRegistration reg) { + this.registration = reg; + } + + @Override + public CompletableResultCode forceFlush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public io.opentelemetry.sdk.metrics.data.AggregationTemporality getAggregationTemporality( + io.opentelemetry.sdk.metrics.InstrumentType instrumentType) { + return io.opentelemetry.sdk.metrics.data.AggregationTemporality.CUMULATIVE; + } + + private Collection collect() { + return this.registration.collectAllMetrics(); + } + } +}