Add simple latency histogram metrics #89
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,25 +1,45 @@ | ||
| package com.salesforce.mirus.metrics; | ||
|
|
||
| import com.google.common.collect.Sets; | ||
| import com.salesforce.mirus.MirusSourceConnector; | ||
| import java.util.*; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.stream.Collectors; | ||
| import org.apache.kafka.common.MetricNameTemplate; | ||
| import org.apache.kafka.common.TopicPartition; | ||
| import org.apache.kafka.common.metrics.MetricConfig; | ||
| import org.apache.kafka.common.metrics.Metrics; | ||
| import org.apache.kafka.common.metrics.Sensor; | ||
| import org.apache.kafka.common.metrics.stats.*; | ||
| import org.apache.kafka.common.utils.Time; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class MirrorJmxReporter extends AbstractMirusJmxReporter { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(MirrorJmxReporter.class); | ||
|
|
||
| public static final Map<Long, String> LATENCY_BUCKETS = | ||
| Map.of( | ||
| TimeUnit.MINUTES.toMillis(0), | ||
| "0m", | ||
| TimeUnit.MINUTES.toMillis(5), | ||
| "5m", | ||
| TimeUnit.MINUTES.toMillis(10), | ||
| "10m", | ||
| TimeUnit.MINUTES.toMillis(30), | ||
| "30m", | ||
| TimeUnit.MINUTES.toMillis(60), | ||
| "60m", | ||
| TimeUnit.HOURS.toMillis(12), | ||
| "12h"); | ||
|
|
||
| private static MirrorJmxReporter instance = null; | ||
|
|
||
| private static final String SOURCE_CONNECTOR_GROUP = MirusSourceConnector.class.getSimpleName(); | ||
|
|
||
| private static final Set<String> TOPIC_TAGS = new HashSet<>(Collections.singletonList("topic")); | ||
| private static final Set<String> TOPIC_BUCKET_TAGS = Sets.newHashSet("topic", "bucket"); | ||
|
|
||
| private static final MetricNameTemplate REPLICATION_LATENCY = | ||
| new MetricNameTemplate( | ||
|
|
@@ -38,16 +58,25 @@ public class MirrorJmxReporter extends AbstractMirusJmxReporter { | |
| "replication-latency-ms-avg", SOURCE_CONNECTOR_GROUP, | ||
| "Average time it takes records to replicate from source to target cluster.", TOPIC_TAGS); | ||
|
|
||
| protected static final MetricNameTemplate HISTOGRAM_LATENCY = | ||
| new MetricNameTemplate( | ||
| "replication-latency-histogram", | ||
| SOURCE_CONNECTOR_GROUP, | ||
| "Cumulative histogram counting records delivered per second with latency exceeding a set of fixed bucket thresholds.", | ||
| TOPIC_BUCKET_TAGS); | ||
|
|
||
| // Map of topics to their metric objects | ||
| private final Map<String, Sensor> topicSensors; | ||
| private final Set<TopicPartition> topicPartitionSet; | ||
| private final Map<String, TreeMap<Long, Sensor>> histogramLatencySensors; | ||
|
|
||
| private MirrorJmxReporter() { | ||
| super(new Metrics(new MetricConfig(), new ArrayList<>(0), Time.SYSTEM, true)); | ||
| metrics.sensor("replication-latency"); | ||
|
|
||
| topicSensors = new HashMap<>(); | ||
| topicPartitionSet = new HashSet<>(); | ||
| histogramLatencySensors = new HashMap<>(); | ||
|
|
||
| logger.info("Initialized MirrorJMXReporter"); | ||
| } | ||
|
|
@@ -73,6 +102,15 @@ public synchronized void addTopics(List<TopicPartition> topicPartitions) { | |
| .filter(topic -> !topicSensors.containsKey(topic)) | ||
| .collect(Collectors.toMap(topic -> topic, this::createTopicSensor))); | ||
| topicPartitionSet.addAll(topicPartitions); | ||
|
|
||
| for (TopicPartition topicPartition : topicPartitions) { | ||
| TreeMap<Long, Sensor> bucketSensors = new TreeMap<>(); | ||
| String topic = topicPartition.topic(); | ||
| LATENCY_BUCKETS.forEach( | ||
| (edgeMillis, bucketName) -> | ||
| bucketSensors.put(edgeMillis, createHistogramSensor(topic, bucketName))); | ||
| histogramLatencySensors.put(topic, bucketSensors); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -104,6 +142,7 @@ public synchronized void removeTopics(List<TopicPartition> topicPartitions) { | |
| topic -> { | ||
| metrics.removeSensor(replicationLatencySensorName(topic)); | ||
| topicSensors.remove(topic); | ||
| histogramLatencySensors.remove(topic); | ||
| }); | ||
| } | ||
|
|
||
|
|
@@ -112,6 +151,24 @@ public synchronized void recordMirrorLatency(String topic, long millis) { | |
| if (sensor != null) { | ||
| sensor.record((double) millis); | ||
| } | ||
|
|
||
| TreeMap<Long, Sensor> bucketSensors = histogramLatencySensors.get(topic); | ||
| for (Map.Entry<Long, Sensor> sensorEntry : bucketSensors.entrySet()) { | ||
| long edgeMillis = sensorEntry.getKey(); | ||
| Sensor bucketSensor = sensorEntry.getValue(); | ||
| if (millis >= edgeMillis) { | ||
| if (bucketSensor.hasExpired()) { | ||
| String bucket = LATENCY_BUCKETS.get(edgeMillis); | ||
| // explicitly replace the expired sensor with a new one | ||
| metrics.removeSensor(histogramLatencySensorName(topic, bucket)); | ||
| bucketSensor = createHistogramSensor(topic, bucket); | ||
| } | ||
| bucketSensor.record(1); | ||
| } else { | ||
| // bucket sensors are sorted by edgeMillis | ||
| break; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private Sensor createTopicSensor(String topic) { | ||
|
|
@@ -127,7 +184,32 @@ private Sensor createTopicSensor(String topic) { | |
| return sensor; | ||
| } | ||
|
|
||
| private Sensor createHistogramSensor(String topic, String bucket) { | ||
| Map<String, String> tags = new LinkedHashMap<>(); | ||
| tags.put("topic", topic); | ||
| tags.put("bucket", bucket); | ||
|
|
||
| // bucket sensor will be expired after 5 mins if inactive | ||
| // this is to prevent inactive bucket sensors from reporting too many zero value metrics | ||
| Sensor sensor = | ||
| metrics.sensor( | ||
| histogramLatencySensorName(topic, bucket), | ||
| null, | ||
| TimeUnit.MINUTES.toSeconds(5), | ||
| Sensor.RecordingLevel.INFO, | ||
| null); | ||
| sensor.add( | ||
| metrics.metricInstance(HISTOGRAM_LATENCY, tags), | ||
| new Rate(TimeUnit.SECONDS, new WindowedSum())); | ||
|
|
||
| return sensor; | ||
| } | ||
|
|
||
| private String replicationLatencySensorName(String topic) { | ||
| return topic + "-" + "replication-latency"; | ||
| } | ||
|
|
||
| private String histogramLatencySensorName(String topic, String bucket) { | ||
| return topic + "-" + bucket + "-" + "histogram-latency"; | ||
|
There was a problem hiding this comment. I find the order strange here, we generally go from less to more specific in naming. How about There was a problem hiding this comment. I think this name is align with the sensor pattern in S3 codebase which has topic and connector. There was a problem hiding this comment. @YongGang OK, I guess the sensor name doesn't matter too much anyway - the sensor names and tags look good. |
||
| } | ||
| } | ||
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,59 @@ | ||
| package com.salesforce.mirus.metrics; | ||
|
|
||
| import java.util.LinkedHashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import org.apache.kafka.common.TopicPartition; | ||
| import org.apache.kafka.common.metrics.Metrics; | ||
| import org.junit.Assert; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
|
|
||
| public class MirrorJmxReporterTest { | ||
|
|
||
| private MirrorJmxReporter mirrorJmxReporter; | ||
| private Metrics metrics; | ||
| private final String TEST_TOPIC = "TestTopic"; | ||
|
|
||
| @Before | ||
| public void setUp() throws Exception { | ||
| mirrorJmxReporter = MirrorJmxReporter.getInstance(); | ||
| metrics = mirrorJmxReporter.metrics; | ||
| } | ||
|
|
||
| @Test | ||
| public void updateLatencyMetrics() { | ||
| TopicPartition topicPartition = new TopicPartition(TEST_TOPIC, 1); | ||
| mirrorJmxReporter.addTopics(List.of(topicPartition)); | ||
|
|
||
| mirrorJmxReporter.recordMirrorLatency(TEST_TOPIC, 500); | ||
|
There was a problem hiding this comment. Have you considered adding a comment here, to declare in which buckets these entries are expected to land? |
||
|
|
||
| Map<String, String> tags = new LinkedHashMap<>(); | ||
| tags.put("topic", TEST_TOPIC); | ||
| tags.put("bucket", "0m"); | ||
| Object value = | ||
| metrics | ||
| .metrics() | ||
| .get( | ||
| metrics.metricName( | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.name(), | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.group(), | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.description(), | ||
| tags)) | ||
| .metricValue(); | ||
| Assert.assertTrue((double) value > 0); | ||
|
|
||
| tags.put("bucket", "12h"); | ||
| value = | ||
| metrics | ||
| .metrics() | ||
| .get( | ||
| metrics.metricName( | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.name(), | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.group(), | ||
| MirrorJmxReporter.HISTOGRAM_LATENCY.description(), | ||
| tags)) | ||
| .metricValue(); | ||
| Assert.assertTrue((double) value == 0); | ||
| } | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we use
TimeUnit.MINUTES.toMillis(60)to be more explicit about units? By convention, milliseconds are usually stored as Longs anyway.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
updated
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It would be nice to make this configurable, but I think that's something we could come back to.