*/
package org.onap.dcae.collectors.veshv.kafkaconsumer.metrics
+import arrow.syntax.function.memoize
+import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.Timer
import io.micrometer.prometheus.PrometheusConfig
import io.micrometer.prometheus.PrometheusMeterRegistry
+import org.apache.kafka.common.TopicPartition
import org.onap.dcae.collectors.veshv.utils.TimeUtils
import reactor.core.publisher.Mono
import java.time.Duration
private val registry: PrometheusMeterRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
) : Metrics {
- private val currentOffset = registry.gauge(name("consumer.offset"), AtomicLong(0))
- private val travelTime = Timer.builder(name("travel.time"))
+ private val currentOffsetByTopicPartition = { topic: String ->
+ registry.gauge(name(CONSUMER, OFFSET, TOPIC), listOf(Tag.of(TOPIC, topic)), AtomicLong(0))
+ }.memoize<String, AtomicLong>()
+
+ private val travelTime = Timer.builder(name(TRAVEL,TIME))
.publishPercentileHistogram(true)
.register(registry)
registry.scrape()
}
- override fun notifyOffsetChanged(offset: Long, topic: String, partition: Int) {
- // TODO use topic and partition
- currentOffset.lazySet(offset)
+ override fun notifyOffsetChanged(offset: Long, topicPartition: TopicPartition) {
+ currentOffsetByTopicPartition(topicPartition.toString()).set(offset)
}
override fun notifyMessageTravelTime(messageSentTimeMicros: Long) {
}
companion object {
- val INSTANCE by lazy { MicrometerMetrics() }
+ val INSTANCE by lazy { MicrometerMetrics() }
+ private const val CONSUMER = "consumer"
+ private const val OFFSET = "offset"
+ private const val TOPIC = "topic"
+ private const val TRAVEL = "travel"
+ private const val TIME = "time"
private const val PREFIX = "hv-kafka-consumer"
private fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"
}
fun update(topicPartition: TopicPartition, offset: Long) {
logger.trace {
- "Current consumer offset $offset for topic ${topicPartition.topic()} " +
- "on partition ${topicPartition.partition()}"
+ "Current consumer offset $offset for topic partition $topicPartition"
}
- metrics.notifyOffsetChanged(offset, topicPartition.topic(), topicPartition.partition())
+ metrics.notifyOffsetChanged(offset, topicPartition)
}
- fun reset() = Unit
-
companion object {
val logger = Logger(OffsetConsumer::class)
}
import io.micrometer.prometheus.PrometheusConfig
import io.micrometer.prometheus.PrometheusMeterRegistry
+import org.apache.kafka.common.TopicPartition
import org.assertj.core.api.Assertions.assertThat
import org.assertj.core.data.Percentage
import org.jetbrains.spek.api.Spek
}
describe("Gauges") {
- val gaugeName = "$PREFIX.consumer.offset"
+ val gaugeName = "$PREFIX.consumer.offset.topic"
on("notifyOffsetChanged") {
val offset = 966L
+ val topicPartition = TopicPartition("sample_topic", 1)
it("should update $gaugeName") {
- cut.notifyOffsetChanged(offset, "sample_topic", 1)
+ cut.notifyOffsetChanged(offset, topicPartition)
registry.verifyGauge(gaugeName) {
assertThat(it.value()).isCloseTo(offset.toDouble(), doublePrecision)