diff options
author | 2019-07-03 13:31:31 +0200 | |
---|---|---|
committer | 2019-07-03 13:31:31 +0200 | |
commit | 867deb3761cdf9c02a2140b16ad42d8f593dc2c9 (patch) | |
tree | 169e7d2ecaa5d65511195b561e30d8790748348a /sources/hv-collector-kafka-consumer/src/main | |
parent | 3e4f108d66fb31e5ea713fef1cf7bc99c0510c21 (diff) |
Change handling of offset metric for consumer app
Change-Id: Ia83469783d64d0174624489df671006e72bc8948
Issue-ID: DCAEGEN2-1635
Signed-off-by: kjaniak <kornel.janiak@nokia.com>
Diffstat (limited to 'sources/hv-collector-kafka-consumer/src/main')
3 files changed, 21 insertions, 12 deletions
diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt index e576a88f..55fae457 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt @@ -19,7 +19,9 @@ */ package org.onap.dcae.collectors.veshv.kafkaconsumer.metrics +import org.apache.kafka.common.TopicPartition + interface Metrics { - fun notifyOffsetChanged(offset: Long, topic: String, partition: Int = 0) + fun notifyOffsetChanged(offset: Long, topicPartition: TopicPartition) fun notifyMessageTravelTime(messageSentTimeMicros: Long) } diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt index da1225e9..0af2cb22 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt @@ -19,9 +19,12 @@ */ 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 @@ -32,8 +35,11 @@ internal class MicrometerMetrics constructor( 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) @@ -41,9 +47,8 @@ internal class MicrometerMetrics constructor( 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) { @@ -51,8 +56,13 @@ internal class MicrometerMetrics constructor( } 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(".")}" } diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt index 1481a224..57a5f33f 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt @@ -28,14 +28,11 @@ internal class OffsetConsumer(private val metrics: Metrics) { 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) } |