-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: prometheus metrics exporter, add new mandatory field server.id t…
…o metronome (#1200) Co-authored-by: Mijail Rondon <[email protected]>
- Loading branch information
Showing
16 changed files
with
216 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
131 changes: 131 additions & 0 deletions
131
canary/src/test/java/io/littlehorse/canary/aggregator/internal/MetricStoreExporterTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package io.littlehorse.canary.aggregator.internal; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.mockito.ArgumentMatchers.any; | ||
| import static org.mockito.Mockito.doNothing; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import io.littlehorse.canary.proto.MetricKey; | ||
| import io.littlehorse.canary.proto.MetricValue; | ||
| import io.littlehorse.canary.proto.Tag; | ||
| import io.micrometer.prometheusmetrics.PrometheusConfig; | ||
| import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; | ||
| import java.time.Duration; | ||
| import java.util.List; | ||
| import org.apache.kafka.streams.KafkaStreams; | ||
| import org.apache.kafka.streams.KeyValue; | ||
| import org.apache.kafka.streams.state.KeyValueIterator; | ||
| import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; | ||
| import org.junit.jupiter.api.AfterEach; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.ExtendWith; | ||
| import org.mockito.Mock; | ||
| import org.mockito.junit.jupiter.MockitoExtension; | ||
|
|
||
| @ExtendWith(MockitoExtension.class) | ||
| class MetricStoreExporterTest { | ||
|
|
||
| public static final String TEST_STORAGE = "testStorage"; | ||
| public static final String HOST = "localhost"; | ||
|
|
||
| @Mock | ||
| KafkaStreams kafkaStreams; | ||
|
|
||
| @Mock | ||
| ReadOnlyKeyValueStore<MetricKey, MetricValue> store; | ||
|
|
||
| @Mock | ||
| KeyValueIterator<MetricKey, MetricValue> records; | ||
|
|
||
| PrometheusMeterRegistry prometheusRegistry; | ||
| MetricStoreExporter metricExporter; | ||
|
|
||
| @BeforeEach | ||
| void setUp() { | ||
| metricExporter = new MetricStoreExporter(kafkaStreams, TEST_STORAGE, Duration.ofSeconds(10)); | ||
| prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); | ||
| } | ||
|
|
||
| @AfterEach | ||
| void tearDown() throws InterruptedException { | ||
| metricExporter.close(); | ||
| prometheusRegistry.close(); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldScrapeSimpleMetric() throws InterruptedException { | ||
| // metrics | ||
| MetricKey key = createMetricsKey(List.of( | ||
| Tag.newBuilder().setKey("custom_tag").setValue("custom_value").build())); | ||
| MetricKey key2 = createMetricsKey(List.of()); | ||
| MetricValue value = MetricValue.newBuilder().setValue(1.0).build(); | ||
|
|
||
| // records | ||
| when(records.hasNext()).thenReturn(true, false); | ||
| when(records.next()).thenReturn(KeyValue.pair(key, value)); | ||
| doNothing().when(records).close(); | ||
|
|
||
| // store | ||
| when(store.all()).thenReturn(records); | ||
|
|
||
| // kafka streams | ||
| when(kafkaStreams.state()).thenReturn(KafkaStreams.State.RUNNING); | ||
| when(kafkaStreams.store(any())).thenReturn(store); | ||
|
|
||
| metricExporter.bindTo(prometheusRegistry); | ||
|
|
||
| Thread.sleep(500); | ||
|
|
||
| assertThat(prometheusRegistry.scrape()) | ||
| .contains( | ||
| "my_metric{custom_tag=\"custom_value\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0"); | ||
| } | ||
|
|
||
| private static MetricKey createMetricsKey(List<Tag> tags) { | ||
| return createMetricsKey(HOST, tags); | ||
| } | ||
|
|
||
| private static MetricKey createMetricsKey(String host, List<Tag> tags) { | ||
| return MetricKey.newBuilder() | ||
| .setServerHost(host) | ||
| .setServerPort(2023) | ||
| .setServerVersion("test") | ||
| .setId("my_metric") | ||
| .setServerId("my_server") | ||
| .addAllTags(tags) | ||
| .build(); | ||
| } | ||
|
|
||
| @Test | ||
| void printMetricsWithTwoDifferentServers() throws InterruptedException { | ||
| // metrics | ||
| List<Tag> tags = List.of( | ||
| Tag.newBuilder().setKey("custom_tag").setValue("custom_value").build()); | ||
| MetricKey key1 = createMetricsKey(tags); | ||
| MetricKey key2 = createMetricsKey("localhost2", tags); | ||
| MetricValue value = MetricValue.newBuilder().setValue(1.0).build(); | ||
|
|
||
| // records | ||
| when(records.hasNext()).thenReturn(true, true, false); | ||
| when(records.next()).thenReturn(KeyValue.pair(key1, value), KeyValue.pair(key2, value)); | ||
| doNothing().when(records).close(); | ||
|
|
||
| // store | ||
| when(store.all()).thenReturn(records); | ||
|
|
||
| // kafka streams | ||
| when(kafkaStreams.state()).thenReturn(KafkaStreams.State.RUNNING); | ||
| when(kafkaStreams.store(any())).thenReturn(store); | ||
|
|
||
| metricExporter.bindTo(prometheusRegistry); | ||
|
|
||
| Thread.sleep(500); | ||
| System.out.printf(prometheusRegistry.scrape()); | ||
| assertThat(prometheusRegistry.scrape()) | ||
| .isEqualTo( | ||
| "# HELP my_metric \n" + "# TYPE my_metric gauge\n" | ||
| + "my_metric{custom_tag=\"custom_value\",server=\"localhost2:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n" | ||
| + "my_metric{custom_tag=\"custom_value\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n"); | ||
| } | ||
| } |
Oops, something went wrong.