Add graphs for performance monitoring

Performance tests need graphs representing: total, message queue on HV-VES,
offset per kafka topic partition, incoming data rate. Added deletion of producers at test end.
Gauges for offsets turned to Range vector (distinguishable based on tag).
Since Producer is to fast it was reconfigured to send messages with 1 ms interval.

Change-Id: I09fc1edbcd824294d0bb89acc2ea2306a6df00ca
Issue-ID: DCAEGEN2-1807
Signed-off-by: kjaniak <kornel.janiak@nokia.com>

3 files changed, 16 insertions, 11 deletions

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 da6a4676..906fce23 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
@@ -36,7 +36,7 @@ internal class MicrometerMetrics constructor(
 ) : Metrics {
 
     private val currentOffsetByTopicPartition = { topicPartition: String ->
-        registry.gauge(name(OFFSET, PARTITION, topicPartition.toLowerCase()),
+        registry.gauge(name(OFFSET, PARTITION),
                 listOf(Tag.of(PARTITION, topicPartition)),
                 AtomicLong(0))
     }.memoize<String, AtomicLong>()
diff --git a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
index cfe67df2..fec7e54e 100644
--- a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
+++ b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
@@ -69,33 +69,32 @@ object MicrometerMetricsTest : Spek({
     }
 
     describe("Gauges") {
-        val gaugeName1 = "$PREFIX.offset.partition.sample_topic-0"
-        val gaugeName2 = "$PREFIX.offset.partition.sample_topic-1"
+        val gaugeName = "$PREFIX.offset.partition"
         val offset1 = 966L
         val offset2 = 967L
         val topicPartition1 = TopicPartition("sample_topic", 0)
         val topicPartition2 = TopicPartition("sample_topic", 1)
 
         on("notifyOffsetChanged") {
-            it("should update $gaugeName1") {
+            it("should update $gaugeName") {
                 cut.notifyOffsetChanged(offset1, topicPartition1)
 
-                registry.verifyGauge(gaugeName1) {
+                registry.verifyGauge(name = gaugeName, tagValue = topicPartition1.toString()) {
                     assertThat(it.value()).isCloseTo(offset1.toDouble(), doublePrecision)
                 }
             }
         }
 
         on("two partition update") {
-            it("should update $gaugeName1") {
+            it("should update $gaugeName") {
                 cut.notifyOffsetChanged(offset1, topicPartition1)
                 cut.notifyOffsetChanged(offset2, topicPartition2)
 
-                registry.verifyGauge(gaugeName1) {
+                registry.verifyGauge(name = gaugeName, tagValue = topicPartition1.toString()) {
                     assertThat(it.value()).isCloseTo(offset1.toDouble(), doublePrecision)
                 }
 
-                registry.verifyGauge(gaugeName2) {
+                registry.verifyGauge(name = gaugeName, tagValue = topicPartition2.toString()) {
                     assertThat(it.value()).isCloseTo(offset2.toDouble(), doublePrecision)
                 }
             }
diff --git a/sources/hv-collector-test-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/tests/utils/metrics.kt b/sources/hv-collector-test-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/tests/utils/metrics.kt
index 1aefdb34..52f9ae87 100644
--- a/sources/hv-collector-test-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/tests/utils/metrics.kt
+++ b/sources/hv-collector-test-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/tests/utils/metrics.kt
@@ -32,6 +32,9 @@ import org.assertj.core.api.Assertions
 fun <T> PrometheusMeterRegistry.verifyGauge(name: String, verifier: (Gauge) -> T) =
         verifyMeter(findMeter(name), RequiredSearch::gauge, verifier)
 
+fun <T> PrometheusMeterRegistry.verifyGauge(name: String, tagKey: String = "partition", tagValue: String, verifier: (Gauge) -> T) =
+        verifyMeter(findMeter(name, tagKey, tagValue), RequiredSearch::gauge, verifier)
+
 fun <T> PrometheusMeterRegistry.verifyTimer(name: String, verifier: (Timer) -> T) =
         verifyMeter(findMeter(name), RequiredSearch::timer, verifier)
 
@@ -43,13 +46,16 @@ fun <T> PrometheusMeterRegistry.verifyCounter(name: String, tags: Tags, verifier
 
 private fun PrometheusMeterRegistry.findMeter(meterName: String) = RequiredSearch.`in`(this).name(meterName)
 
+private fun PrometheusMeterRegistry.findMeter(meterName: String, tagKey: String, tagValue: String) =
+        RequiredSearch.`in`(this).tag(tagKey, tagValue).name(meterName)
+
 private fun <T> verifyCounter(search: RequiredSearch, verifier: (Counter) -> T) =
         verifyMeter(search, RequiredSearch::counter, verifier)
 
 private inline fun <M, T> verifyMeter(search: RequiredSearch,
-                               map: (RequiredSearch) -> M,
-                               verifier: (M) -> T) =
+                                      map: (RequiredSearch) -> M,
+                                      verifier: (M) -> T) =
         Try { map(search) }.fold(
                 { ex -> Assertions.assertThat(ex).doesNotThrowAnyException() },
                 verifier
-        )
\ No newline at end of file
+        )