Message travel time: Producer -> HV-VES input introduced.
Tests for new metric added.
Change-Id: I36347ff53abb3f274e4358af26db49fe8bac95ed
Issue-ID: DCAEGEN2-1576
Signed-off-by: kjaniak <kornel.janiak@nokia.com>
interface Metrics {
fun notifyBytesReceived(size: Int)
fun notifyMessageReceived(msg: WireFrameMessage)
+ fun notifyMessageReceived(msg: VesMessage)
fun notifyMessageReadyForRouting(msg: VesMessage)
fun notifyMessageSent(msg: RoutedMessage)
fun notifyMessageDropped(cause: MessageDropCause)
* ============LICENSE_START=======================================================
* dcaegen2-collectors-veshv
* ================================================================================
- * Copyright (C) 2018-2019 NOKIA
+ * Copyright (C) 2018-2020 NOKIA
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
.filterFailedWithLog(logger, clientContext::fullMdc,
{ "Ves event header decoded successfully" },
{ "Failed to decode ves event header, reason: ${it.message}" })
- }
+ }.doOnNext(metrics::notifyMessageReceived)
private fun filterInvalidProtobufMessages(flux: Flux<VesMessage>): Flux<VesMessage> = flux
.filterFailedWithLog {
* ============LICENSE_START=======================================================
* dcaegen2-collectors-veshv
* ================================================================================
- * Copyright (C) 2018-2019 NOKIA
+ * Copyright (C) 2018-2020 NOKIA
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import org.onap.dcae.collectors.veshv.model.MessageDropCause
import org.onap.dcae.collectors.veshv.domain.RoutedMessage
import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.dcae.collectors.veshv.utils.TimeUtils
import java.time.Duration
import java.time.Instant
import kotlin.test.fail
var messagesDroppedCount: Int = 0; private set
var lastProcessingTimeMicros: Double = -1.0; private set
var lastProcessingTimeWithoutRoutingMicros: Double = -1.0; private set
+ var lastToCollectorTravelTime: Double = -1.0; private set
var messagesSentCount: Int = 0; private set
var clientRejectionCause = mutableMapOf<ClientRejectionCause, Int>(); private set
messageBytesReceived += msg.payloadSize
}
+ override fun notifyMessageReceived(msg: VesMessage) {
+ lastToCollectorTravelTime = Duration.between(TimeUtils.epochMicroToInstant(msg.header.lastEpochMicrosec),
+ Instant.now()).toNanos() / 1000.0
+ }
+
override fun notifyMessageReadyForRouting(msg: VesMessage) {
lastProcessingTimeWithoutRoutingMicros = Duration.between(msg.wtpFrame.receivedAt, Instant.now()).toNanos() / 1000.0
}
* ============LICENSE_START=======================================================
* dcaegen2-collectors-veshv
* ================================================================================
- * Copyright (C) 2018-2019 NOKIA
+ * Copyright (C) 2018-2020 NOKIA
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import io.micrometer.prometheus.PrometheusConfig
import io.micrometer.prometheus.PrometheusMeterRegistry
import org.onap.dcae.collectors.veshv.boundary.Metrics
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
import org.onap.dcae.collectors.veshv.model.MessageDropCause
-import org.onap.dcae.collectors.veshv.domain.RoutedMessage
-import org.onap.dcae.collectors.veshv.domain.VesMessage
import org.onap.dcae.collectors.veshv.utils.TimeUtils.epochMicroToInstant
import java.time.Duration
import java.time.Instant
private val totalConnections = registry.counter(name(CONNECTIONS))
private val disconnections = registry.counter(name(DISCONNECTIONS))
+ private val travelTimeToCollector = Timer.builder(name(MESSAGES, TO, COLLECTOR, TRAVEL, TIME))
+ .maximumExpectedValue(MAX_BUCKET_DURATION)
+ .publishPercentileHistogram(true)
+ .register(registry)
private val processingTime = Timer.builder(name(MESSAGES, PROCESSING, TIME))
.maximumExpectedValue(MAX_BUCKET_DURATION)
.publishPercentileHistogram(true)
receivedMessagesPayloadBytes.increment(msg.payloadSize.toDouble())
}
+ override fun notifyMessageReceived(msg: VesMessage) {
+ travelTimeToCollector.record(
+ Duration.between(epochMicroToInstant(msg.header.lastEpochMicrosec), msg.wtpFrame.receivedAt)
+ )
+ }
+
override fun notifyMessageSent(msg: RoutedMessage) {
val now = Instant.now()
sentMessages.increment()
internal const val PAYLOAD = "payload"
internal const val WITHOUT = "without"
internal const val ROUTING = "routing"
+ internal const val TRAVEL = "travel"
+ internal const val TO = "to"
+ internal const val COLLECTOR = "collector"
internal val MAX_BUCKET_DURATION = Duration.ofSeconds(300L)
internal fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"
}
* ============LICENSE_START=======================================================
* dcaegen2-collectors-veshv
* ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2018-2020 NOKIA
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import org.jetbrains.spek.api.dsl.describe
import org.jetbrains.spek.api.dsl.it
import org.jetbrains.spek.api.dsl.on
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesEventDomain
+import org.onap.dcae.collectors.veshv.domain.VesMessage
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.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.domain.RoutedMessage
-import org.onap.dcae.collectors.veshv.domain.VesEventDomain
-import org.onap.dcae.collectors.veshv.domain.VesMessage
import org.onap.dcae.collectors.veshv.tests.utils.commonHeader
import org.onap.dcae.collectors.veshv.tests.utils.emptyWireProtocolFrame
import org.onap.dcae.collectors.veshv.tests.utils.verifyCounter
}
}
+ on("$PREFIX.messages.to.collector.travel.time") {
+ val counterName = "$PREFIX.messages.to.collector.travel.time"
+ val toCollectorTravelTimeMs = 100L
+
+ it("should update timer") {
+ val now = Instant.now()
+ val vesMessage = vesMessageReceivedAt(now, sentAt = now.minusMillis(toCollectorTravelTimeMs))
+ cut.notifyMessageReceived(vesMessage)
+
+ registry.verifyTimer(counterName) { timer ->
+ assertThat(timer.mean(TimeUnit.MILLISECONDS)).isEqualTo(toCollectorTravelTimeMs.toDouble())
+ }
+
+ verifyCountersAndTimersAreUnchangedBut(counterName)
+ }
+ }
+
on("$PREFIX.messages.processing.time.without.routing") {
val counterName = "$PREFIX.messages.processing.time.without.routing"
val processingTimeMs = 100L
registry.verifyTimer(counterName) { timer ->
assertThat(timer.mean(TimeUnit.MILLISECONDS)).isGreaterThanOrEqualTo(processingTimeMs.toDouble())
}
- verifyCountersAndTimersAreUnchangedBut(
- counterName,
- "$PREFIX.messages.sent.topic",
- "$PREFIX.messages.sent",
- "$PREFIX.messages.latency")
+
+ verifyCountersAndTimersAreUnchangedBut(counterName)
}
}
}
})
+private fun vesMessageReceivedAt(receivedAt: Instant, sentAt: Instant): VesMessage {
+ val lastEpochMicrosec = sentAt.epochSecond * 1000000 + sentAt.nano / 1000
+ val commonHeader = commonHeader(lastEpochMicrosec = lastEpochMicrosec)
+ return VesMessage(commonHeader,
+ wireProtocolFrame(commonHeader, ByteString.copyFromUtf8("highvolume measurements"), receivedAt))
+}
+
private fun vesMessageReceivedAt(receivedAt: Temporal, domain: VesEventDomain = VesEventDomain.PERF3GPP): VesMessage {
val commonHeader = commonHeader(domain)
return VesMessage(commonHeader,