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(); + } + } +}