import io.micrometer.prometheus.PrometheusMeterRegistry
import org.onap.dcae.collectors.veshv.boundary.Metrics
import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
-import org.onap.dcae.collectors.veshv.model.MessageDropCause
import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
+import org.onap.dcae.collectors.veshv.model.MessageDropCause
import org.onap.dcae.collectors.veshv.model.RoutedMessage
+import org.onap.dcae.collectors.veshv.utils.TimeUtils.epochMicroToInstant
import java.time.Duration
import java.time.Instant
private val receivedMsgCount = registry.counter(name(MESSAGES, RECEIVED, COUNT))
private val receivedMsgBytes = registry.counter(name(MESSAGES, RECEIVED, BYTES))
+ private val processingTime = registry.timer(name(MESSAGES, PROCESSING, TIME))
+ private val totalLatency = registry.timer(name(MESSAGES, LATENCY, TIME))
+
private val sentCountTotal = registry.counter(name(MESSAGES, SENT, COUNT))
private val sentToTopicCount = { topic: String ->
registry.counter(name(MESSAGES, SENT, TOPIC, COUNT), TOPIC, topic)
registry.counter(name(MESSAGES, DROPPED, CAUSE, COUNT), CAUSE, cause)
}.memoize<String, Counter>()
- private val processingTime = registry.timer(name(MESSAGES, PROCESSING, TIME))
-
private val clientsRejectedCount = registry.counter(name(CLIENTS, REJECTED, COUNT))
private val clientsRejectedCauseCount = { cause: String ->
registry.counter(name(CLIENTS, REJECTED, CAUSE, COUNT), CAUSE, cause)
}
override fun notifyMessageSent(msg: RoutedMessage) {
+ val now = Instant.now()
sentCountTotal.increment()
sentToTopicCount(msg.topic).increment()
- processingTime.record(Duration.between(msg.message.wtpFrame.receivedAt, Instant.now()))
+
+ processingTime.record(Duration.between(msg.message.wtpFrame.receivedAt, now))
+ totalLatency.record(Duration.between(epochMicroToInstant(msg.message.header.lastEpochMicrosec), now))
}
override fun notifyMessageDropped(cause: MessageDropCause) {
internal const val TOPIC = "topic"
internal const val DROPPED = "dropped"
internal const val TIME = "time"
- fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"
+ internal const val LATENCY = "latency"
+ internal fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"
}
}
import arrow.core.Try
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Gauge
+import io.micrometer.core.instrument.Meter
import io.micrometer.core.instrument.Timer
import io.micrometer.core.instrument.search.RequiredSearch
import io.micrometer.prometheus.PrometheusConfig
import org.jetbrains.spek.api.dsl.on
import org.onap.dcae.collectors.veshv.main.metrics.MicrometerMetrics
import org.onap.dcae.collectors.veshv.main.metrics.MicrometerMetrics.Companion.PREFIX
-import org.onap.dcae.collectors.veshv.model.MessageDropCause.INVALID_MESSAGE
-import org.onap.dcae.collectors.veshv.model.MessageDropCause.ROUTE_NOT_FOUND
import org.onap.dcae.collectors.veshv.model.ClientRejectionCause.INVALID_WIRE_FRAME_MARKER
import org.onap.dcae.collectors.veshv.model.ClientRejectionCause.PAYLOAD_SIZE_EXCEEDED_IN_MESSAGE
+import org.onap.dcae.collectors.veshv.model.MessageDropCause.INVALID_MESSAGE
+import org.onap.dcae.collectors.veshv.model.MessageDropCause.ROUTE_NOT_FOUND
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.dcae.collectors.veshv.tests.utils.emptyWireProtocolFrame
import java.time.Instant
import java.time.temporal.Temporal
import java.util.concurrent.TimeUnit
+import kotlin.reflect.KClass
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
*/
object MicrometerMetricsTest : Spek({
val doublePrecision = Percentage.withPercentage(0.5)
+ val alwaysChangedMeters = setOf("$PREFIX.messages.processing.time", "$PREFIX.messages.latency.time")
lateinit var registry: PrometheusMeterRegistry
lateinit var cut: MicrometerMetrics
fun <T> verifyCounter(name: String, verifier: (Counter) -> T) =
verifyCounter(registrySearch(name), verifier)
- fun verifyAllCountersAreUnchangedBut(vararg changedCounters: String) {
- registry.meters
- .filter { it.id.name.startsWith(PREFIX) }
- .filter { it is Counter }
- .map { it as Counter }
- .filterNot { it.id.name in changedCounters }
- .forEach {
- assertThat(it.count()).describedAs(it.id.toString()).isCloseTo(0.0, doublePrecision)
- }
+ fun verifyCountersAndTimersAreUnchangedBut(vararg changedMeters: String) {
+ fun <T : Meter> verifyAllMetersAreUnchangedBut(
+ clazz: KClass<T>,
+ changedCounters: Collection<String>,
+ valueOf: (T) -> Double) {
+ registry.meters
+ .filter { it.id.name.startsWith(PREFIX) }
+ .filter { clazz.isInstance(it) }
+ .map { it as T }
+ .filterNot { it.id.name in changedCounters }
+ .forEach {
+ assertThat(valueOf(it)).describedAs(it.id.toString()).isCloseTo(0.0, doublePrecision)
+ }
+ }
+
+ setOf(*changedMeters).let { changedMetersCollection ->
+ verifyAllMetersAreUnchangedBut(Counter::class, changedMetersCollection) { it.count() }
+ verifyAllMetersAreUnchangedBut(Timer::class, changedMetersCollection) { it.count().toDouble() }
+ }
}
describe("notifyBytesReceived") {
it("should leave all other counters unchanged") {
cut.notifyBytesReceived(128)
- verifyAllCountersAreUnchangedBut(counterName)
+ verifyCountersAndTimersAreUnchangedBut(counterName)
}
}
}
it("should leave all other counters unchanged") {
cut.notifyMessageReceived(emptyWireProtocolFrame().copy(payloadSize = 128))
- verifyAllCountersAreUnchangedBut(
+ verifyCountersAndTimersAreUnchangedBut(
"$PREFIX.messages.received.count",
"$PREFIX.messages.received.bytes"
)
verifyCounter(counterName) {
assertThat(it.count()).isCloseTo(1.0, doublePrecision)
}
- verifyAllCountersAreUnchangedBut(counterName, "$PREFIX.messages.sent.topic.count")
+ verifyCountersAndTimersAreUnchangedBut(
+ counterName,
+ "$PREFIX.messages.sent.topic.count",
+ "$PREFIX.messages.processing.time",
+ "$PREFIX.messages.latency.time")
}
}
it("should update timer") {
- cut.notifyMessageSent(routedMessage(topicName1, Instant.now().minusMillis(processingTimeMs)))
+ cut.notifyMessageSent(routedMessageReceivedAt(topicName1, Instant.now().minusMillis(processingTimeMs)))
verifyTimer(counterName) { timer ->
assertThat(timer.mean(TimeUnit.MILLISECONDS)).isGreaterThanOrEqualTo(processingTimeMs.toDouble())
}
- verifyAllCountersAreUnchangedBut(
+ verifyCountersAndTimersAreUnchangedBut(
counterName,
"$PREFIX.messages.sent.topic.count",
- "$PREFIX.messages.sent.count")
+ "$PREFIX.messages.sent.count",
+ "$PREFIX.messages.latency.time")
}
}
+
+ on("$PREFIX.messages.latency.time") {
+ val counterName = "$PREFIX.messages.latency.time"
+ val latencyMs = 1666L
+
+ it("should update timer") {
+
+ cut.notifyMessageSent(routedMessageSentAt(topicName1, Instant.now().minusMillis(latencyMs)))
+
+ verifyTimer(counterName) { timer ->
+ assertThat(timer.mean(TimeUnit.MILLISECONDS))
+ .isGreaterThanOrEqualTo(latencyMs.toDouble())
+ .isLessThanOrEqualTo(latencyMs + 10000.0)
+
+ }
+ verifyCountersAndTimersAreUnchangedBut(
+ counterName,
+ "$PREFIX.messages.sent.topic.count",
+ "$PREFIX.messages.sent.count",
+ "$PREFIX.messages.processing.time")
+ }
+ }
+
}
describe("notifyMessageDropped") {
verifyCounter(counterName) {
assertThat(it.count()).isCloseTo(2.0, doublePrecision)
}
- verifyAllCountersAreUnchangedBut(counterName, "$PREFIX.messages.dropped.cause.count")
+ verifyCountersAndTimersAreUnchangedBut(counterName, "$PREFIX.messages.dropped.cause.count")
}
}
verifyCounter(counterName) {
assertThat(it.count()).isCloseTo(2.0, doublePrecision)
}
- verifyAllCountersAreUnchangedBut(counterName, "$PREFIX.clients.rejected.cause.count")
+ verifyCountersAndTimersAreUnchangedBut(counterName, "$PREFIX.clients.rejected.cause.count")
}
}
})
fun routedMessage(topic: String, partition: Int = 0) =
- vesEvent().let {evt ->
+ vesEvent().let { evt ->
RoutedMessage(topic, partition,
VesMessage(evt.commonEventHeader, wireProtocolFrame(evt)))
}
-fun routedMessage(topic: String, receivedAt: Temporal, partition: Int = 0) =
- vesEvent().let {evt ->
+fun routedMessageReceivedAt(topic: String, receivedAt: Temporal, partition: Int = 0) =
+ vesEvent().let { evt ->
RoutedMessage(topic, partition,
VesMessage(evt.commonEventHeader, wireProtocolFrame(evt).copy(receivedAt = receivedAt)))
}
+
+fun routedMessageSentAt(topic: String, sentAt: Instant, partition: Int = 0) =
+ vesEvent().let { evt ->
+ val builder = evt.toBuilder()
+ builder.commonEventHeaderBuilder.lastEpochMicrosec = sentAt.epochSecond * 1000000 + sentAt.nano / 1000
+ builder.build()
+ }.let { evt ->
+ RoutedMessage(topic, partition,
+ VesMessage(evt.commonEventHeader, wireProtocolFrame(evt)))
+ }
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.collectors.veshv.utils
+
+import java.time.Instant
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since December 2018
+ */
+object TimeUtils {
+ fun epochMicroToInstant(epochMicroseconds: Long): Instant {
+ val seconds = epochMicroseconds / MICROSECONDS_IN_SECOND
+ val nanos = (epochMicroseconds - seconds * MICROSECONDS_IN_SECOND) * NANOSECONDS_IN_MICROSECOND
+ return Instant.ofEpochSecond(seconds, nanos)
+ }
+
+ private const val MICROSECONDS_IN_SECOND = 1_000_000L
+ private const val NANOSECONDS_IN_MICROSECOND = 1_000L
+}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.collectors.veshv.utils
+
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.it
+import java.time.Instant
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since December 2018
+ */
+internal object TimeTest : Spek({
+ describe("epochMicrosecond to Instant converter") {
+ it("should convert") {
+ val epochSeconds = 1545048422L
+ val nanoAdjustment = 666999000L
+ val epochMicros = 1545048422666999L
+
+ val result = TimeUtils.epochMicroToInstant(epochMicros)
+
+ assertThat(result).isEqualTo(Instant.ofEpochSecond(epochSeconds, nanoAdjustment))
+ }
+ }
+})
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff