refactor(metrics): Make MetricUtils.registry non-nullable

- Make MetricUtils.registry non-nullable with default no-op implementation. This allows us to remove boilerplate for handling the null case
- Rename request context metric names to follow convention
This commit is contained in:
Abe 2025-09-09 11:08:16 -07:00 committed by GitHub
parent 4ea758da19
commit 6ec6f0150d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
27 changed files with 191 additions and 396 deletions

View File

@ -87,13 +87,9 @@ public class GraphQLEngine {
graphQLQueryComplexityLimit, new DataHubFieldComplexityCalculator())); graphQLQueryComplexityLimit, new DataHubFieldComplexityCalculator()));
if (metricUtils != null && graphQLConfiguration.getMetrics().isEnabled()) { if (metricUtils != null && graphQLConfiguration.getMetrics().isEnabled()) {
metricUtils
.getRegistry()
.ifPresent(
meterRegistry ->
instrumentations.add( instrumentations.add(
new GraphQLTimingInstrumentation( new GraphQLTimingInstrumentation(
meterRegistry, graphQLConfiguration.getMetrics()))); metricUtils.getRegistry(), graphQLConfiguration.getMetrics()));
} }
ChainedInstrumentation chainedInstrumentation = new ChainedInstrumentation(instrumentations); ChainedInstrumentation chainedInstrumentation = new ChainedInstrumentation(instrumentations);

View File

@ -284,7 +284,7 @@ public class GraphQLEngineTest {
public void testMetricsEnabled() { public void testMetricsEnabled() {
// Setup metrics // Setup metrics
MeterRegistry meterRegistry = new SimpleMeterRegistry(); MeterRegistry meterRegistry = new SimpleMeterRegistry();
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Enable metrics in configuration // Enable metrics in configuration
GraphQLConfiguration metricsEnabledConfig = createDefaultConfiguration(); GraphQLConfiguration metricsEnabledConfig = createDefaultConfiguration();
@ -405,7 +405,7 @@ public class GraphQLEngineTest {
metrics.setTrivialDataFetchersEnabled(true); metrics.setTrivialDataFetchersEnabled(true);
MeterRegistry meterRegistry = new SimpleMeterRegistry(); MeterRegistry meterRegistry = new SimpleMeterRegistry();
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
graphQLEngine = graphQLEngine =
GraphQLEngine.builder() GraphQLEngine.builder()
@ -457,7 +457,7 @@ public class GraphQLEngineTest {
@Test @Test
public void testMetricsWithEmptyRegistry() { public void testMetricsWithEmptyRegistry() {
// Setup metrics with empty registry // Setup metrics with empty registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.empty()); when(mockMetricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
GraphQLConfiguration metricsEnabledConfig = createDefaultConfiguration(); GraphQLConfiguration metricsEnabledConfig = createDefaultConfiguration();
metricsEnabledConfig.getMetrics().setEnabled(true); metricsEnabledConfig.getMetrics().setEnabled(true);

View File

@ -124,7 +124,7 @@ public class ElasticSearchTimeseriesAspectService
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
if (metricUtils != null) { if (metricUtils != null) {
MicrometerMetricsRegistry.registerExecutorMetrics( MicrometerMetricsRegistry.registerExecutorMetrics(
"timeseries", this.queryPool, metricUtils.getRegistry().orElse(null)); "timeseries", this.queryPool, metricUtils.getRegistry());
} }
this.entityRegistry = entityRegistry; this.entityRegistry = entityRegistry;

View File

@ -30,7 +30,6 @@ import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import java.util.Collections; import java.util.Collections;
import java.util.Map; import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.MockitoAnnotations; import org.mockito.MockitoAnnotations;
@ -97,7 +96,7 @@ public class GraphQLTimingInstrumentationTest {
// Setup real GraphQL configuration for GraphQLEngine // Setup real GraphQL configuration for GraphQLEngine
graphQLConfiguration = createDefaultConfiguration(); graphQLConfiguration = createDefaultConfiguration();
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
} }
@Test @Test

View File

@ -69,9 +69,6 @@ public abstract class AbstractKafkaListener<E, H extends EventHook<E>, R>
// TODO: include priority level when available // TODO: include priority level when available
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer( .timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME, MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic", "topic",
@ -80,7 +77,6 @@ public abstract class AbstractKafkaListener<E, H extends EventHook<E>, R>
consumerGroupId) consumerGroupId)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
}); });
});
final R record = consumerRecord.value(); final R record = consumerRecord.value();
log.debug( log.debug(
"Got event consumer: {} key: {}, topic: {}, partition: {}, offset: {}, value size: {}, timestamp: {}", "Got event consumer: {} key: {}, topic: {}, partition: {}, offset: {}, value size: {}, timestamp: {}",

View File

@ -76,9 +76,6 @@ public class DataHubUsageEventsProcessor {
// TODO: include priority level when available // TODO: include priority level when available
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer( .timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME, MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic", "topic",
@ -87,7 +84,6 @@ public class DataHubUsageEventsProcessor {
datahubUsageEventConsumerGroupId) datahubUsageEventConsumerGroupId)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
}); });
});
final String record = consumerRecord.value(); final String record = consumerRecord.value();
log.info( log.info(

View File

@ -118,12 +118,8 @@ public class MCLKafkaListener
// request // request
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer(MetricUtils.DATAHUB_REQUEST_HOOK_QUEUE_TIME, "hook", hookName) .timer(MetricUtils.DATAHUB_REQUEST_HOOK_QUEUE_TIME, "hook", hookName)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
});
} }
}); });
} }

View File

@ -49,6 +49,8 @@ public class DataHubUsageEventsProcessorTest {
public void setUp() { public void setUp() {
MockitoAnnotations.openMocks(this); MockitoAnnotations.openMocks(this);
when(metricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
systemOperationContext = systemOperationContext =
TestOperationContexts.Builder.builder() TestOperationContexts.Builder.builder()
.systemTelemetryContextSupplier( .systemTelemetryContextSupplier(
@ -65,6 +67,8 @@ public class DataHubUsageEventsProcessorTest {
dataHubUsageEventTransformer, dataHubUsageEventTransformer,
systemOperationContext.getSearchContext().getIndexConvention(), systemOperationContext.getSearchContext().getIndexConvention(),
systemOperationContext); systemOperationContext);
// Set the consumer group ID
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
} }
@Test @Test
@ -346,10 +350,7 @@ public class DataHubUsageEventsProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(metricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID via reflection
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
// Setup test data // Setup test data
String eventJson = "{\"type\":\"PageViewEvent\",\"timestamp\":1234567890}"; String eventJson = "{\"type\":\"PageViewEvent\",\"timestamp\":1234567890}";
@ -397,10 +398,7 @@ public class DataHubUsageEventsProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(metricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
// Setup test data // Setup test data
String eventJson = "{\"type\":\"SearchEvent\"}"; String eventJson = "{\"type\":\"SearchEvent\"}";
@ -466,10 +464,7 @@ public class DataHubUsageEventsProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(metricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
// Setup test data // Setup test data
String eventJson = "{\"invalid\":\"event\"}"; String eventJson = "{\"invalid\":\"event\"}";
@ -504,89 +499,13 @@ public class DataHubUsageEventsProcessorTest {
verify(elasticsearchConnector, never()).feedElasticEvent(any()); verify(elasticsearchConnector, never()).feedElasticEvent(any());
} }
@Test
public void testMicrometerMetricsAbsentWhenRegistryNotPresent() {
// Configure the mock metricUtils to return empty Optional (no registry)
when(metricUtils.getRegistry()).thenReturn(Optional.empty());
// Set the consumer group ID
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
// Setup test data
String eventJson = "{\"type\":\"ViewEvent\"}";
String transformedDocument = "{\"view\":\"data\"}";
when(mockRecord.timestamp()).thenReturn(System.currentTimeMillis() - 1000);
when(mockRecord.value()).thenReturn(eventJson);
DataHubUsageEventTransformer.TransformedDocument transformedDoc =
new DataHubUsageEventTransformer.TransformedDocument(TEST_EVENT_ID, transformedDocument);
when(dataHubUsageEventTransformer.transformDataHubUsageEvent(eventJson))
.thenReturn(Optional.of(transformedDoc));
// Execute - should not throw exception
processor.consume(mockRecord);
// Verify the histogram method was still called (for dropwizard metrics)
verify(metricUtils).histogram(eq(DataHubUsageEventsProcessor.class), eq("kafkaLag"), anyLong());
// Verify processing completed successfully despite no registry
verify(elasticsearchConnector).feedElasticEvent(any());
}
@Test
public void testMicrometerKafkaQueueTimeWithCustomConsumerGroup() {
// Setup a real MeterRegistry
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry));
// Set a custom consumer group ID
String customConsumerGroup = "custom-usage-event-consumer";
setConsumerGroupId(processor, customConsumerGroup);
// Setup test data
String eventJson = "{\"type\":\"Event\"}";
String transformedDocument = "{\"data\":\"value\"}";
when(mockRecord.timestamp()).thenReturn(System.currentTimeMillis() - 1500);
when(mockRecord.topic()).thenReturn("DataHubUsageEvent_v1");
when(mockRecord.value()).thenReturn(eventJson);
DataHubUsageEventTransformer.TransformedDocument transformedDoc =
new DataHubUsageEventTransformer.TransformedDocument(TEST_EVENT_ID, transformedDocument);
when(dataHubUsageEventTransformer.transformDataHubUsageEvent(eventJson))
.thenReturn(Optional.of(transformedDoc));
// Execute
processor.consume(mockRecord);
// Verify timer was recorded with custom consumer group
Timer timer =
meterRegistry.timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic",
"DataHubUsageEvent_v1",
"consumer.group",
customConsumerGroup);
assertNotNull(timer);
assertEquals(timer.count(), 1);
}
@Test @Test
public void testMicrometerKafkaQueueTimeAccuracy() { public void testMicrometerKafkaQueueTimeAccuracy() {
// Setup a real MeterRegistry // Setup a real MeterRegistry
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(metricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID
setConsumerGroupId(processor, "datahub-usage-event-consumer-job-client");
// Setup test data // Setup test data
String eventJson = "{\"type\":\"Event\"}"; String eventJson = "{\"type\":\"Event\"}";

View File

@ -64,7 +64,7 @@ public class MCLKafkaListenerTest {
mockSystemMetadata = spy(SystemMetadataUtils.createDefaultSystemMetadata()); mockSystemMetadata = spy(SystemMetadataUtils.createDefaultSystemMetadata());
meterRegistry = new SimpleMeterRegistry(); meterRegistry = new SimpleMeterRegistry();
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(metricUtils.getRegistry()).thenReturn(meterRegistry);
systemOperationContext = systemOperationContext =
TestOperationContexts.Builder.builder() TestOperationContexts.Builder.builder()

View File

@ -93,9 +93,6 @@ public class MetadataChangeProposalsProcessor {
// TODO: include priority level when available // TODO: include priority level when available
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer( .timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME, MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic", "topic",
@ -104,7 +101,6 @@ public class MetadataChangeProposalsProcessor {
mceConsumerGroupId) mceConsumerGroupId)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
}); });
});
final GenericRecord record = consumerRecord.value(); final GenericRecord record = consumerRecord.value();
log.info( log.info(

View File

@ -95,9 +95,6 @@ public class BatchMetadataChangeProposalsProcessor {
// TODO: include priority level when available // TODO: include priority level when available
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer( .timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME, MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic", "topic",
@ -106,7 +103,6 @@ public class BatchMetadataChangeProposalsProcessor {
mceConsumerGroupId) mceConsumerGroupId)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
}); });
});
final GenericRecord record = consumerRecord.value(); final GenericRecord record = consumerRecord.value();
if (topicName == null) { if (topicName == null) {

View File

@ -1,7 +1,6 @@
package com.linkedin.metadata.kafka; package com.linkedin.metadata.kafka;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.mockStatic;
@ -41,6 +40,7 @@ import com.linkedin.mxe.Topics;
import io.datahubproject.metadata.context.OperationContext; import io.datahubproject.metadata.context.OperationContext;
import io.datahubproject.metadata.context.SystemTelemetryContext; import io.datahubproject.metadata.context.SystemTelemetryContext;
import io.datahubproject.test.metadata.context.TestOperationContexts; import io.datahubproject.test.metadata.context.TestOperationContexts;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer; import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.Span;
@ -48,7 +48,6 @@ import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.Tracer;
import java.io.IOException; import java.io.IOException;
import java.util.List; import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecord;
@ -100,18 +99,18 @@ public class MetadataChangeProposalsProcessorTest {
@Mock private Span mockSpan; @Mock private Span mockSpan;
@Mock private MetricUtils metricUtils; private MetricUtils metricUtils;
private AutoCloseable mocks; private AutoCloseable mocks;
private MockedStatic<Span> spanMock; private MockedStatic<Span> spanMock;
private MockedStatic<MetricUtils> metricUtilsMock;
private MockedStatic<EventUtils> eventUtilsMock; private MockedStatic<EventUtils> eventUtilsMock;
@BeforeMethod @BeforeMethod
public void setup() { public void setup() {
mocks = MockitoAnnotations.openMocks(this); mocks = MockitoAnnotations.openMocks(this);
metricUtils = MetricUtils.builder().registry(new SimpleMeterRegistry()).build();
opContext = opContext =
opContext.toBuilder() opContext.toBuilder()
.systemTelemetryContext( .systemTelemetryContext(
@ -160,11 +159,6 @@ public class MetadataChangeProposalsProcessorTest {
spanMock = mockStatic(Span.class); spanMock = mockStatic(Span.class);
spanMock.when(Span::current).thenReturn(mockSpan); spanMock.when(Span::current).thenReturn(mockSpan);
metricUtilsMock = mockStatic(MetricUtils.class);
metricUtilsMock
.when(() -> MetricUtils.name(eq(MetadataChangeProposalsProcessor.class), any()))
.thenReturn("metricName");
eventUtilsMock = mockStatic(EventUtils.class); eventUtilsMock = mockStatic(EventUtils.class);
// Setup consumer record mock // Setup consumer record mock
@ -185,11 +179,6 @@ public class MetadataChangeProposalsProcessorTest {
spanMock = null; // Set to null after closing spanMock = null; // Set to null after closing
} }
if (metricUtilsMock != null) {
metricUtilsMock.close();
metricUtilsMock = null; // Set to null after closing
}
if (eventUtilsMock != null) { if (eventUtilsMock != null) {
eventUtilsMock.close(); eventUtilsMock.close();
eventUtilsMock = null; // Set to null after closing eventUtilsMock = null; // Set to null after closing
@ -344,11 +333,7 @@ public class MetadataChangeProposalsProcessorTest {
@Test @Test
public void testMicrometerKafkaQueueTimeMetric() throws Exception { public void testMicrometerKafkaQueueTimeMetric() throws Exception {
// Setup a real MeterRegistry MeterRegistry meterRegistry = metricUtils.getRegistry();
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry));
// Set timestamp to simulate queue time // Set timestamp to simulate queue time
long messageTimestamp = System.currentTimeMillis() - 3000; // 3 seconds ago long messageTimestamp = System.currentTimeMillis() - 3000; // 3 seconds ago
@ -376,21 +361,13 @@ public class MetadataChangeProposalsProcessorTest {
assertTrue(timer.totalTime(TimeUnit.MILLISECONDS) >= 2500); // At least 2.5 seconds assertTrue(timer.totalTime(TimeUnit.MILLISECONDS) >= 2500); // At least 2.5 seconds
assertTrue(timer.totalTime(TimeUnit.MILLISECONDS) <= 3500); // At most 3.5 seconds assertTrue(timer.totalTime(TimeUnit.MILLISECONDS) <= 3500); // At most 3.5 seconds
// Verify the histogram method was called
verify(metricUtils)
.histogram(eq(MetadataChangeProposalsProcessor.class), eq("kafkaLag"), anyLong());
// Verify successful processing // Verify successful processing
verify(mockEntityService).ingestProposal(eq(opContext), any(), eq(false)); verify(mockEntityService).ingestProposal(eq(opContext), any(), eq(false));
} }
@Test @Test
public void testMicrometerKafkaQueueTimeWithDifferentTopics() throws Exception { public void testMicrometerKafkaQueueTimeWithDifferentTopics() throws Exception {
// Setup a real MeterRegistry MeterRegistry meterRegistry = metricUtils.getRegistry();
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry));
// Create MCP // Create MCP
MetadataChangeProposal mcp = createSimpleMCP(); MetadataChangeProposal mcp = createSimpleMCP();
@ -448,11 +425,7 @@ public class MetadataChangeProposalsProcessorTest {
@Test @Test
public void testMicrometerMetricsWithProcessingFailure() throws Exception { public void testMicrometerMetricsWithProcessingFailure() throws Exception {
// Setup a real MeterRegistry MeterRegistry meterRegistry = metricUtils.getRegistry();
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry));
// Create MCP that will fail // Create MCP that will fail
MetadataChangeProposal mcp = new MetadataChangeProposal(); MetadataChangeProposal mcp = new MetadataChangeProposal();
@ -492,9 +465,6 @@ public class MetadataChangeProposalsProcessorTest {
@Test @Test
public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception { public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception {
// Configure the mock metricUtils to return empty Optional (no registry)
when(metricUtils.getRegistry()).thenReturn(Optional.empty());
// Create MCP // Create MCP
MetadataChangeProposal mcp = createSimpleMCP(); MetadataChangeProposal mcp = createSimpleMCP();
eventUtilsMock.when(() -> EventUtils.avroToPegasusMCP(mockRecord)).thenReturn(mcp); eventUtilsMock.when(() -> EventUtils.avroToPegasusMCP(mockRecord)).thenReturn(mcp);
@ -504,21 +474,13 @@ public class MetadataChangeProposalsProcessorTest {
// Execute - should not throw exception // Execute - should not throw exception
processor.consume(mockConsumerRecord); processor.consume(mockConsumerRecord);
// Verify the histogram method was still called (for dropwizard metrics)
verify(metricUtils)
.histogram(eq(MetadataChangeProposalsProcessor.class), eq("kafkaLag"), anyLong());
// Verify processing completed successfully despite no registry // Verify processing completed successfully despite no registry
verify(mockEntityService).ingestProposal(eq(opContext), any(), eq(false)); verify(mockEntityService).ingestProposal(eq(opContext), any(), eq(false));
} }
@Test @Test
public void testMicrometerKafkaQueueTimeAccuracy() throws Exception { public void testMicrometerKafkaQueueTimeAccuracy() throws Exception {
// Setup a real MeterRegistry MeterRegistry meterRegistry = metricUtils.getRegistry();
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry
when(metricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry));
// Create MCP // Create MCP
MetadataChangeProposal mcp = createSimpleMCP(); MetadataChangeProposal mcp = createSimpleMCP();
@ -562,10 +524,6 @@ public class MetadataChangeProposalsProcessorTest {
// Verify max recorded time // Verify max recorded time
assertTrue(timer.max(TimeUnit.MILLISECONDS) >= 4500); assertTrue(timer.max(TimeUnit.MILLISECONDS) >= 4500);
assertTrue(timer.max(TimeUnit.MILLISECONDS) <= 5500); assertTrue(timer.max(TimeUnit.MILLISECONDS) <= 5500);
// Verify histogram was called for each record
verify(metricUtils, times(queueTimes.length))
.histogram(eq(MetadataChangeProposalsProcessor.class), eq("kafkaLag"), anyLong());
} }
@Test @Test
@ -598,10 +556,6 @@ public class MetadataChangeProposalsProcessorTest {
// Verify processing completed successfully // Verify processing completed successfully
verify(mockEntityService).ingestProposal(eq(opContextNoMetrics), any(), eq(false)); verify(mockEntityService).ingestProposal(eq(opContextNoMetrics), any(), eq(false));
// Verify metricUtils methods were never called since it's not present in context
verify(metricUtils, never()).histogram(any(), any(), anyLong());
verify(metricUtils, never()).getRegistry();
} }
// Helper method // Helper method

View File

@ -557,6 +557,7 @@ public class BatchMetadataChangeProposalsProcessorTest {
public void testConsumeWithTelemetryMetrics() throws Exception { public void testConsumeWithTelemetryMetrics() throws Exception {
// Mock the metric utils // Mock the metric utils
MetricUtils mockMetricUtils = mock(MetricUtils.class); MetricUtils mockMetricUtils = mock(MetricUtils.class);
when(mockMetricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
// Create a mock operation context // Create a mock operation context
OperationContext opContextWithMetrics = spy(opContext); OperationContext opContextWithMetrics = spy(opContext);
@ -856,12 +857,12 @@ public class BatchMetadataChangeProposalsProcessorTest {
@Test @Test
public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception { public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception {
// Create MetricUtils without a registry // With the new non-null contract, we can't create MetricUtils with null registry
MetricUtils metricUtilsNoRegistry = MetricUtils.builder().registry(null).build(); // Instead, test the case where MetricUtils is absent from OperationContext
// Create operation context with metric utils that has no registry // Create operation context without metric utils
OperationContext opContextNoRegistry = mock(OperationContext.class); OperationContext opContextNoRegistry = mock(OperationContext.class);
when(opContextNoRegistry.getMetricUtils()).thenReturn(Optional.of(metricUtilsNoRegistry)); when(opContextNoRegistry.getMetricUtils()).thenReturn(Optional.empty());
// Mock withQueueSpan // Mock withQueueSpan
doAnswer( doAnswer(

View File

@ -82,9 +82,6 @@ public class PlatformEventProcessor {
// TODO: include priority level when available // TODO: include priority level when available
metricUtils metricUtils
.getRegistry() .getRegistry()
.ifPresent(
meterRegistry -> {
meterRegistry
.timer( .timer(
MetricUtils.KAFKA_MESSAGE_QUEUE_TIME, MetricUtils.KAFKA_MESSAGE_QUEUE_TIME,
"topic", "topic",
@ -93,7 +90,6 @@ public class PlatformEventProcessor {
datahubPlatformEventConsumerGroupId) datahubPlatformEventConsumerGroupId)
.record(Duration.ofMillis(queueTimeMs)); .record(Duration.ofMillis(queueTimeMs));
}); });
});
final GenericRecord record = consumerRecord.value(); final GenericRecord record = consumerRecord.value();
log.info( log.info(
"Got PE event key: {}, topic: {}, partition: {}, offset: {}, value size: {}, timestamp: {}", "Got PE event key: {}, topic: {}, partition: {}, offset: {}, value size: {}, timestamp: {}",

View File

@ -62,6 +62,7 @@ public class PlatformEventProcessorTest {
// Create mock MetricUtils // Create mock MetricUtils
mockMetricUtils = mock(MetricUtils.class); mockMetricUtils = mock(MetricUtils.class);
when(mockMetricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
when(mockOperationContext.getMetricUtils()).thenReturn(Optional.of(mockMetricUtils)); when(mockOperationContext.getMetricUtils()).thenReturn(Optional.of(mockMetricUtils));
// Create mock hooks // Create mock hooks
@ -133,6 +134,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2); List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
try (MockedStatic<EventUtils> mockedEventUtils = Mockito.mockStatic(EventUtils.class)) { try (MockedStatic<EventUtils> mockedEventUtils = Mockito.mockStatic(EventUtils.class)) {
mockedEventUtils mockedEventUtils
@ -161,6 +164,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1); List<PlatformEventHook> hooks = Arrays.asList(mockHook1);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
try (MockedStatic<EventUtils> mockedEventUtils = Mockito.mockStatic(EventUtils.class)) { try (MockedStatic<EventUtils> mockedEventUtils = Mockito.mockStatic(EventUtils.class)) {
mockedEventUtils mockedEventUtils
@ -189,6 +194,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2); List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
// Make first hook throw exception // Make first hook throw exception
doThrow(new RuntimeException("Hook 1 failed")) doThrow(new RuntimeException("Hook 1 failed"))
@ -224,6 +231,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2); List<PlatformEventHook> hooks = Arrays.asList(mockHook1, mockHook2);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
// Make both hooks throw exceptions // Make both hooks throw exceptions
doThrow(new RuntimeException("Hook 1 failed")) doThrow(new RuntimeException("Hook 1 failed"))
@ -283,6 +292,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1); List<PlatformEventHook> hooks = Arrays.asList(mockHook1);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
// Set up specific consumer record values // Set up specific consumer record values
String expectedKey = "test-key-123"; String expectedKey = "test-key-123";
@ -321,7 +332,8 @@ public class PlatformEventProcessorTest {
// Verify that the consumer record methods were called // Verify that the consumer record methods were called
// Note: some methods may be called multiple times (e.g., for logging and metrics) // Note: some methods may be called multiple times (e.g., for logging and metrics)
verify(specificMockRecord, times(1)).key(); verify(specificMockRecord, times(1)).key();
verify(specificMockRecord, times(1)).topic(); verify(specificMockRecord, times(2))
.topic(); // Called twice: once for metrics, once for logging
verify(specificMockRecord, times(1)).partition(); verify(specificMockRecord, times(1)).partition();
verify(specificMockRecord, times(1)).offset(); verify(specificMockRecord, times(1)).offset();
verify(specificMockRecord, times(2)) verify(specificMockRecord, times(2))
@ -336,6 +348,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1); List<PlatformEventHook> hooks = Arrays.asList(mockHook1);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
PlatformEvent event1 = mock(PlatformEvent.class); PlatformEvent event1 = mock(PlatformEvent.class);
when(event1.getName()).thenReturn("Event1"); when(event1.getName()).thenReturn("Event1");
@ -389,6 +403,8 @@ public class PlatformEventProcessorTest {
// Setup // Setup
List<PlatformEventHook> hooks = Arrays.asList(mockHook1); List<PlatformEventHook> hooks = Arrays.asList(mockHook1);
processor = new PlatformEventProcessor(mockOperationContext, hooks); processor = new PlatformEventProcessor(mockOperationContext, hooks);
// Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client");
ConsumerRecord<String, GenericRecord> nullValueRecord = mock(ConsumerRecord.class); ConsumerRecord<String, GenericRecord> nullValueRecord = mock(ConsumerRecord.class);
when(nullValueRecord.value()).thenReturn(null); when(nullValueRecord.value()).thenReturn(null);
@ -422,7 +438,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID via reflection // Set the consumer group ID via reflection
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");
@ -469,7 +485,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID // Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");
@ -534,7 +550,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID // Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");
@ -578,7 +594,7 @@ public class PlatformEventProcessorTest {
@Test @Test
public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception { public void testMicrometerMetricsAbsentWhenRegistryNotPresent() throws Exception {
// Configure the mock metricUtils to return empty Optional (no registry) // Configure the mock metricUtils to return empty Optional (no registry)
when(mockMetricUtils.getRegistry()).thenReturn(Optional.empty()); when(mockMetricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
// Set the consumer group ID // Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");
@ -608,7 +624,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set a custom consumer group ID // Set a custom consumer group ID
String customConsumerGroup = "custom-platform-event-consumer"; String customConsumerGroup = "custom-platform-event-consumer";
@ -645,7 +661,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID // Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");
@ -705,7 +721,7 @@ public class PlatformEventProcessorTest {
SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
// Configure the mock metricUtils to return the registry // Configure the mock metricUtils to return the registry
when(mockMetricUtils.getRegistry()).thenReturn(Optional.of(meterRegistry)); when(mockMetricUtils.getRegistry()).thenReturn(meterRegistry);
// Set the consumer group ID // Set the consumer group ID
setConsumerGroupId(processor, "generic-platform-event-job-client"); setConsumerGroupId(processor, "generic-platform-event-job-client");

View File

@ -275,11 +275,11 @@ public class RequestContext implements ContextInterface {
metricUtils.incrementMicrometer( metricUtils.incrementMicrometer(
MetricUtils.DATAHUB_REQUEST_COUNT, MetricUtils.DATAHUB_REQUEST_COUNT,
1, 1,
"user.category", "user_category",
userCategory, userCategory,
"agent.class", "agent_class",
agentClass, agentClass,
"request.api", "request_api",
requestAPI); requestAPI);
} }
} }

View File

@ -278,13 +278,13 @@ public class RequestContextTest {
verify(mockMetricUtils, atLeastOnce()) verify(mockMetricUtils, atLeastOnce())
.incrementMicrometer( .incrementMicrometer(
eq("datahub.request.count"), eq(MetricUtils.DATAHUB_REQUEST_COUNT),
eq(1.0d), eq(1.0d),
eq("user.category"), eq("user_category"),
eq("system"), eq("system"),
eq("agent.class"), eq("agent_class"),
eq("unknown"), eq("unknown"),
eq("request.api"), eq("request_api"),
eq("restli")); eq("restli"));
} }
@ -303,13 +303,13 @@ public class RequestContextTest {
verify(mockMetricUtils, atLeastOnce()) verify(mockMetricUtils, atLeastOnce())
.incrementMicrometer( .incrementMicrometer(
eq("datahub.request.count"), eq(MetricUtils.DATAHUB_REQUEST_COUNT),
eq(1.0d), eq(1.0d),
eq("user.category"), eq("user_category"),
eq("admin"), eq("admin"),
eq("agent.class"), eq("agent_class"),
eq("unknown"), eq("unknown"),
eq("request.api"), eq("request_api"),
eq("restli")); eq("restli"));
} }
@ -328,13 +328,13 @@ public class RequestContextTest {
verify(mockMetricUtils, atLeastOnce()) verify(mockMetricUtils, atLeastOnce())
.incrementMicrometer( .incrementMicrometer(
eq("datahub.request.count"), eq(MetricUtils.DATAHUB_REQUEST_COUNT),
eq(1.0d), eq(1.0d),
eq("user.category"), eq("user_category"),
eq("regular"), eq("regular"),
eq("agent.class"), eq("agent_class"),
eq("unknown"), eq("unknown"),
eq("request.api"), eq("request_api"),
eq("restli")); eq("restli"));
} }

View File

@ -317,7 +317,7 @@ public class GraphQLEngineFactory {
GraphQLConcurrencyUtils.setExecutorService(graphQLWorkerPool); GraphQLConcurrencyUtils.setExecutorService(graphQLWorkerPool);
if (metricUtils != null) { if (metricUtils != null) {
MicrometerMetricsRegistry.registerExecutorMetrics( MicrometerMetricsRegistry.registerExecutorMetrics(
"graphql", graphqlExecutorService, metricUtils.getRegistry().orElse(null)); "graphql", graphqlExecutorService, metricUtils.getRegistry());
} }
return graphQLWorkerPool; return graphQLWorkerPool;

View File

@ -86,7 +86,7 @@ public class KafkaTraceReaderFactory {
traceExecutorService = Executors.newFixedThreadPool(threadPoolSize); traceExecutorService = Executors.newFixedThreadPool(threadPoolSize);
if (metricUtils != null) { if (metricUtils != null) {
MicrometerMetricsRegistry.registerExecutorMetrics( MicrometerMetricsRegistry.registerExecutorMetrics(
"api-trace", this.traceExecutorService, metricUtils.getRegistry().orElse(null)); "api-trace", this.traceExecutorService, metricUtils.getRegistry());
} }
return traceExecutorService; return traceExecutorService;
} }

View File

@ -41,6 +41,7 @@ import io.datahubproject.metadata.context.SystemTelemetryContext;
import io.datahubproject.metadata.services.RestrictedService; import io.datahubproject.metadata.services.RestrictedService;
import io.datahubproject.metadata.services.SecretService; import io.datahubproject.metadata.services.SecretService;
import io.datahubproject.test.metadata.context.TestOperationContexts; import io.datahubproject.test.metadata.context.TestOperationContexts;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.Tracer;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
@ -261,7 +262,7 @@ public class GraphQLEngineFactoryTest extends AbstractTestNGSpringContextTests {
public void setUp() { public void setUp() {
// Set up default mock behaviors // Set up default mock behaviors
when(graphService.supportsMultiHop()).thenReturn(true); when(graphService.supportsMultiHop()).thenReturn(true);
when(metricUtils.getRegistry()).thenReturn(java.util.Optional.empty()); when(metricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
} }
@Test @Test

View File

@ -109,7 +109,7 @@ public class ClientCache<K, V, C extends ClientCacheConfig> {
if (config.isStatsEnabled() && metricUtils != null) { if (config.isStatsEnabled() && metricUtils != null) {
MicrometerMetricsRegistry.registerCacheMetrics( MicrometerMetricsRegistry.registerCacheMetrics(
config.getName(), cache, metricUtils.getRegistry().orElse(null)); config.getName(), cache, metricUtils.getRegistry());
} }
return new ClientCache<>(config, cache, loadFunction, weigher, ttlSecondsFunction); return new ClientCache<>(config, cache, loadFunction, weigher, ttlSecondsFunction);

View File

@ -145,7 +145,7 @@ public class RestliEntityClient extends BaseClient implements EntityClient {
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
if (metricUtils != null) { if (metricUtils != null) {
MicrometerMetricsRegistry.registerExecutorMetrics( MicrometerMetricsRegistry.registerExecutorMetrics(
"entity-client-get", this.batchGetV2Pool, metricUtils.getRegistry().orElse(null)); "entity-client-get", this.batchGetV2Pool, metricUtils.getRegistry());
} }
this.batchIngestPool = this.batchIngestPool =
new ThreadPoolExecutor( new ThreadPoolExecutor(
@ -158,7 +158,7 @@ public class RestliEntityClient extends BaseClient implements EntityClient {
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
if (metricUtils != null) { if (metricUtils != null) {
MicrometerMetricsRegistry.registerExecutorMetrics( MicrometerMetricsRegistry.registerExecutorMetrics(
"entity-client-ingest", this.batchIngestPool, metricUtils.getRegistry().orElse(null)); "entity-client-ingest", this.batchIngestPool, metricUtils.getRegistry());
} }
} }

View File

@ -63,6 +63,7 @@ import com.linkedin.restli.client.RestLiResponseException;
import com.linkedin.restli.client.response.BatchKVResponse; import com.linkedin.restli.client.response.BatchKVResponse;
import io.datahubproject.metadata.context.OperationContext; import io.datahubproject.metadata.context.OperationContext;
import io.datahubproject.test.metadata.context.TestOperationContexts; import io.datahubproject.test.metadata.context.TestOperationContexts;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import java.net.URISyntaxException; import java.net.URISyntaxException;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
@ -98,7 +99,7 @@ public class BaseClientTest {
// Setup MetricUtils mock // Setup MetricUtils mock
mockMetricUtils = mock(MetricUtils.class); mockMetricUtils = mock(MetricUtils.class);
when(mockMetricUtils.getRegistry()).thenReturn(Optional.empty()); when(mockMetricUtils.getRegistry()).thenReturn(new SimpleMeterRegistry());
} }
@Test @Test

View File

@ -7,11 +7,12 @@ import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer; import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
import java.util.Map; import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import lombok.Builder; import lombok.Builder;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@Slf4j @Slf4j
@ -21,10 +22,10 @@ public class MetricUtils {
public static final String DROPWIZARD_METRIC = "dwizMetric"; public static final String DROPWIZARD_METRIC = "dwizMetric";
public static final String DROPWIZARD_NAME = "dwizName"; public static final String DROPWIZARD_NAME = "dwizName";
/* Micrometer */ /* Micrometer. See https://prometheus.io/docs/practices/naming/ */
public static final String KAFKA_MESSAGE_QUEUE_TIME = "kafka.message.queue.time"; public static final String KAFKA_MESSAGE_QUEUE_TIME = "kafka.message.queue.time";
public static final String DATAHUB_REQUEST_HOOK_QUEUE_TIME = "datahub.request.hook.queue.time"; public static final String DATAHUB_REQUEST_HOOK_QUEUE_TIME = "datahub.request.hook.queue.time";
public static final String DATAHUB_REQUEST_COUNT = "datahub.request.count"; public static final String DATAHUB_REQUEST_COUNT = "datahub_request_count";
/* OpenTelemetry */ /* OpenTelemetry */
public static final String CACHE_HIT_ATTR = "cache.hit"; public static final String CACHE_HIT_ATTR = "cache.hit";
@ -42,7 +43,7 @@ public class MetricUtils {
@Deprecated public static final String DELIMITER = "_"; @Deprecated public static final String DELIMITER = "_";
private final MeterRegistry registry; @Builder.Default @NonNull private final MeterRegistry registry = new CompositeMeterRegistry();
private static final Map<String, Timer> legacyTimeCache = new ConcurrentHashMap<>(); private static final Map<String, Timer> legacyTimeCache = new ConcurrentHashMap<>();
private static final Map<String, Counter> legacyCounterCache = new ConcurrentHashMap<>(); private static final Map<String, Counter> legacyCounterCache = new ConcurrentHashMap<>();
private static final Map<String, DistributionSummary> legacyHistogramCache = private static final Map<String, DistributionSummary> legacyHistogramCache =
@ -52,24 +53,17 @@ public class MetricUtils {
// For state-based gauges (like throttled state) // For state-based gauges (like throttled state)
private static final Map<String, AtomicDouble> gaugeStates = new ConcurrentHashMap<>(); private static final Map<String, AtomicDouble> gaugeStates = new ConcurrentHashMap<>();
public Optional<MeterRegistry> getRegistry() { public MeterRegistry getRegistry() {
return Optional.ofNullable(registry); return registry;
} }
@Deprecated @Deprecated
public void time(String dropWizardMetricName, long durationNanos) { public void time(String dropWizardMetricName, long durationNanos) {
getRegistry()
.ifPresent(
meterRegistry -> {
Timer timer = Timer timer =
legacyTimeCache.computeIfAbsent( legacyTimeCache.computeIfAbsent(
dropWizardMetricName, dropWizardMetricName,
name -> name -> Timer.builder(name).tags(DROPWIZARD_METRIC, "true").register(registry));
Timer.builder(name)
.tags(DROPWIZARD_METRIC, "true")
.register(meterRegistry));
timer.record(durationNanos, TimeUnit.NANOSECONDS); timer.record(durationNanos, TimeUnit.NANOSECONDS);
});
} }
@Deprecated @Deprecated
@ -90,18 +84,14 @@ public class MetricUtils {
@Deprecated @Deprecated
public void increment(String metricName, double increment) { public void increment(String metricName, double increment) {
getRegistry()
.ifPresent(
meterRegistry -> {
Counter counter = Counter counter =
legacyCounterCache.computeIfAbsent( legacyCounterCache.computeIfAbsent(
metricName, metricName,
name -> name ->
Counter.builder(MetricRegistry.name(name)) Counter.builder(MetricRegistry.name(name))
.tag(DROPWIZARD_METRIC, "true") .tag(DROPWIZARD_METRIC, "true")
.register(meterRegistry)); .register(registry));
counter.increment(increment); counter.increment(increment);
});
} }
/** /**
@ -112,16 +102,11 @@ public class MetricUtils {
* @param tags The tags to associate with the metric (can be empty) * @param tags The tags to associate with the metric (can be empty)
*/ */
public void incrementMicrometer(String metricName, double increment, String... tags) { public void incrementMicrometer(String metricName, double increment, String... tags) {
getRegistry()
.ifPresent(
meterRegistry -> {
// Create a cache key that includes both metric name and tags // Create a cache key that includes both metric name and tags
String cacheKey = createCacheKey(metricName, tags); String cacheKey = createCacheKey(metricName, tags);
Counter counter = Counter counter =
micrometerCounterCache.computeIfAbsent( micrometerCounterCache.computeIfAbsent(cacheKey, key -> registry.counter(metricName, tags));
cacheKey, key -> meterRegistry.counter(metricName, tags));
counter.increment(increment); counter.increment(increment);
});
} }
/** /**
@ -167,9 +152,6 @@ public class MetricUtils {
public void setGaugeValue(Class<?> clazz, String metricName, double value) { public void setGaugeValue(Class<?> clazz, String metricName, double value) {
String name = MetricRegistry.name(clazz, metricName); String name = MetricRegistry.name(clazz, metricName);
getRegistry()
.ifPresent(
meterRegistry -> {
// Get or create the state holder // Get or create the state holder
AtomicDouble state = gaugeStates.computeIfAbsent(name, k -> new AtomicDouble(0)); AtomicDouble state = gaugeStates.computeIfAbsent(name, k -> new AtomicDouble(0));
@ -179,28 +161,21 @@ public class MetricUtils {
key -> key ->
Gauge.builder(key, state, AtomicDouble::get) Gauge.builder(key, state, AtomicDouble::get)
.tag(DROPWIZARD_METRIC, "true") .tag(DROPWIZARD_METRIC, "true")
.register(meterRegistry)); .register(registry));
// Update the value // Update the value
state.set(value); state.set(value);
});
} }
@Deprecated @Deprecated
public void histogram(Class<?> clazz, String metricName, long value) { public void histogram(Class<?> clazz, String metricName, long value) {
getRegistry()
.ifPresent(
meterRegistry -> {
String name = MetricRegistry.name(clazz, metricName); String name = MetricRegistry.name(clazz, metricName);
DistributionSummary summary = DistributionSummary summary =
legacyHistogramCache.computeIfAbsent( legacyHistogramCache.computeIfAbsent(
name, name,
key -> key ->
DistributionSummary.builder(key) DistributionSummary.builder(key).tag(DROPWIZARD_METRIC, "true").register(registry));
.tag(DROPWIZARD_METRIC, "true")
.register(meterRegistry));
summary.record(value); summary.record(value);
});
} }
@Deprecated @Deprecated

View File

@ -27,11 +27,7 @@ public class MicrometerMetricsRegistry {
public static synchronized boolean registerCacheMetrics( public static synchronized boolean registerCacheMetrics(
@Nonnull String cacheName, @Nonnull String cacheName,
@Nonnull Object nativeCache, @Nonnull Object nativeCache,
@Nullable MeterRegistry meterRegistry) { @Nonnull MeterRegistry meterRegistry) {
if (cacheName == null || nativeCache == null || meterRegistry == null) {
return false;
}
if (!GLOBALLY_REGISTERED_CACHES.add(cacheName)) { if (!GLOBALLY_REGISTERED_CACHES.add(cacheName)) {
return false; return false;

View File

@ -10,7 +10,6 @@ import io.micrometer.core.instrument.Meter;
import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer; import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import java.util.Optional;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.function.Supplier; import java.util.function.Supplier;
import org.mockito.Mock; import org.mockito.Mock;
@ -44,18 +43,25 @@ public class MetricUtilsTest {
} }
@Test @Test
public void testGetRegistryReturnsOptionalWithRegistry() { public void testGetRegistryReturnsRegistry() {
Optional<MeterRegistry> registry = metricUtils.getRegistry(); MeterRegistry registry = metricUtils.getRegistry();
assertTrue(registry.isPresent()); assertNotNull(registry);
assertSame(registry.get(), meterRegistry); assertSame(registry, meterRegistry);
} }
@Test @Test
public void testGetRegistryReturnsEmptyOptionalWhenNull() { public void testGetRegistryReturnsDefaultWhenNotSpecified() {
MetricUtils utilsWithNullRegistry = MetricUtils.builder().registry(null).build(); MetricUtils utilsWithDefaultRegistry = MetricUtils.builder().build();
Optional<MeterRegistry> registry = utilsWithNullRegistry.getRegistry(); MeterRegistry registry = utilsWithDefaultRegistry.getRegistry();
assertFalse(registry.isPresent()); assertNotNull(registry);
assertTrue(registry instanceof io.micrometer.core.instrument.composite.CompositeMeterRegistry);
}
@Test(expectedExceptions = NullPointerException.class)
public void testBuilderRejectsNullRegistry() {
// This should throw NullPointerException due to @NonNull annotation
MetricUtils.builder().registry(null).build();
} }
@Test @Test
@ -72,11 +78,11 @@ public class MetricUtilsTest {
} }
@Test @Test
public void testTimeWithNullRegistryDoesNothing() { public void testTimeWithDefaultRegistryWorks() {
MetricUtils utilsWithNullRegistry = MetricUtils.builder().registry(null).build(); MetricUtils utilsWithDefaultRegistry = MetricUtils.builder().build();
// Should not throw exception // Should not throw exception and should work with default registry
utilsWithNullRegistry.time("test.timer", 1000); utilsWithDefaultRegistry.time("test.timer", 1000);
} }
@Test @Test
@ -217,16 +223,21 @@ public class MetricUtilsTest {
} }
@Test @Test
public void testAllMethodsWithNullRegistry() { public void testAllMethodsWithDefaultRegistry() {
MetricUtils utilsWithNullRegistry = MetricUtils.builder().registry(null).build(); MetricUtils utilsWithDefaultRegistry = MetricUtils.builder().build();
// None of these should throw exceptions // All methods should work with the default CompositeMeterRegistry
utilsWithNullRegistry.time("timer", 1000); utilsWithDefaultRegistry.time("timer", 1000);
utilsWithNullRegistry.increment(this.getClass(), "counter", 1); utilsWithDefaultRegistry.increment(this.getClass(), "counter", 1);
utilsWithNullRegistry.increment("counter", 1); utilsWithDefaultRegistry.increment("counter", 1);
utilsWithNullRegistry.exceptionIncrement(this.getClass(), "error", new RuntimeException()); utilsWithDefaultRegistry.exceptionIncrement(this.getClass(), "error", new RuntimeException());
utilsWithNullRegistry.setGaugeValue(this.getClass(), "gauge", 42); utilsWithDefaultRegistry.setGaugeValue(this.getClass(), "gauge", 42);
utilsWithNullRegistry.histogram(this.getClass(), "histogram", 100); utilsWithDefaultRegistry.histogram(this.getClass(), "histogram", 100);
// Verify the registry is the expected type
assertTrue(
utilsWithDefaultRegistry.getRegistry()
instanceof io.micrometer.core.instrument.composite.CompositeMeterRegistry);
} }
@Test @Test

View File

@ -155,32 +155,6 @@ public class MicrometerMetricsRegistryTest {
assertFalse(result); assertFalse(result);
} }
@Test
public void testRegisterCacheWithNullNativeCache() {
// Given
String cacheName = "testCache";
// When
boolean result = MicrometerMetricsRegistry.registerCacheMetrics(cacheName, null, meterRegistry);
// Then
assertFalse(result);
}
@Test
public void testRegisterCacheWithNullCacheName() {
// Given
Cache<Object, Object> caffeineCache =
Caffeine.newBuilder().maximumSize(100).recordStats().build();
// When
boolean result =
MicrometerMetricsRegistry.registerCacheMetrics(null, caffeineCache, meterRegistry);
// Then
assertFalse(result); // Should return false for null cache name
}
@Test @Test
public void testRegisterCacheWithUnsupportedCacheType() { public void testRegisterCacheWithUnsupportedCacheType() {
// Given // Given
@ -352,30 +326,6 @@ public class MicrometerMetricsRegistryTest {
assertEquals(registeredKey, cacheName); assertEquals(registeredKey, cacheName);
} }
@Test
public void testAllNullParameters() {
// When
boolean result = MicrometerMetricsRegistry.registerCacheMetrics(null, null, null);
// Then
assertFalse(result);
}
@Test
public void testNullCheckOrder() {
// Test various combinations of null parameters
// Null cache name
Cache<Object, Object> caffeineCache = Caffeine.newBuilder().maximumSize(100).build();
assertFalse(MicrometerMetricsRegistry.registerCacheMetrics(null, caffeineCache, meterRegistry));
// Null native cache
assertFalse(MicrometerMetricsRegistry.registerCacheMetrics("test", null, meterRegistry));
// Null meter registry
assertFalse(MicrometerMetricsRegistry.registerCacheMetrics("test", caffeineCache, null));
}
@Test @Test
public void testSuccessfulRegistrationDoesNotThrowOnSubsequentAttempts() { public void testSuccessfulRegistrationDoesNotThrowOnSubsequentAttempts() {
// Given // Given