2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2018-2020 NOKIA
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
20 package org.onap.dcae.collectors.veshv.main.metrics
22 import arrow.syntax.function.memoize
23 import io.micrometer.core.instrument.Counter
24 import io.micrometer.core.instrument.Timer
25 import io.micrometer.core.instrument.binder.jvm.ClassLoaderMetrics
26 import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics
27 import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics
28 import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics
29 import io.micrometer.core.instrument.binder.system.ProcessorMetrics
30 import io.micrometer.prometheus.PrometheusConfig
31 import io.micrometer.prometheus.PrometheusMeterRegistry
32 import org.onap.dcae.collectors.veshv.boundary.Metrics
33 import org.onap.dcae.collectors.veshv.domain.RoutedMessage
34 import org.onap.dcae.collectors.veshv.domain.VesMessage
35 import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
36 import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
37 import org.onap.dcae.collectors.veshv.model.MessageDropCause
38 import org.onap.dcae.collectors.veshv.utils.TimeUtils.epochMicroToInstant
39 import java.time.Duration
40 import java.time.Instant
44 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
47 class MicrometerMetrics internal constructor(
48 private val registry: PrometheusMeterRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
50 private val receivedBytes = registry.counter(name(DATA, RECEIVED, BYTES))
51 private val receivedMessages = registry.counter(name(MESSAGES, RECEIVED))
52 private val receivedMessagesPayloadBytes = registry.counter(name(MESSAGES, RECEIVED, PAYLOAD, BYTES))
54 private val totalConnections = registry.counter(name(CONNECTIONS))
55 private val disconnections = registry.counter(name(DISCONNECTIONS))
57 private val travelTimeToCollector = Timer.builder(name(MESSAGES, TO, COLLECTOR, TRAVEL, TIME))
58 .maximumExpectedValue(MAX_BUCKET_DURATION)
59 .publishPercentileHistogram(true)
61 private val processingTime = Timer.builder(name(MESSAGES, PROCESSING, TIME))
62 .maximumExpectedValue(MAX_BUCKET_DURATION)
63 .publishPercentileHistogram(true)
65 private val processingTimeWithoutRouting = Timer.builder(name(MESSAGES, PROCESSING, TIME, WITHOUT, ROUTING))
66 .maximumExpectedValue(MAX_BUCKET_DURATION)
67 .publishPercentileHistogram(true)
69 private val totalLatencyWithoutRouting = Timer.builder(name(MESSAGES, LATENCY, WITHOUT, ROUTING))
70 .maximumExpectedValue(MAX_BUCKET_DURATION)
71 .publishPercentileHistogram(true)
73 private val totalLatency = Timer.builder(name(MESSAGES, LATENCY))
74 .maximumExpectedValue(MAX_BUCKET_DURATION)
75 .publishPercentileHistogram(true)
78 private val sentMessages = registry.counter(name(MESSAGES, SENT))
79 private val sentMessagesByTopic = { topic: String ->
80 registry.counter(name(MESSAGES, SENT, TOPIC), TOPIC, topic)
81 }.memoize<String, Counter>()
82 private val droppedMessages = registry.counter(name(MESSAGES, DROPPED))
83 private val messagesDroppedByCause = { cause: String ->
84 registry.counter(name(MESSAGES, DROPPED, CAUSE), CAUSE, cause)
85 }.memoize<String, Counter>()
86 private val clientsRejected = registry.counter(name(CLIENTS, REJECTED))
87 private val clientsRejectedByCause = { cause: String ->
88 registry.counter(name(CLIENTS, REJECTED, CAUSE), CAUSE, cause)
89 }.memoize<String, Counter>()
93 registry.gauge(name(CONNECTIONS, ACTIVE), this) {
94 (totalConnections.count() - disconnections.count()).coerceAtLeast(0.0)
97 ClassLoaderMetrics().bindTo(registry)
98 JvmMemoryMetrics().bindTo(registry)
99 JvmGcMetrics().bindTo(registry)
100 ProcessorMetrics().bindTo(registry)
101 JvmThreadMetrics().bindTo(registry)
104 val metricsProvider = MicrometerPrometheusMetricsProvider(registry)
106 override fun notifyBytesReceived(size: Int) {
107 receivedBytes.increment(size.toDouble())
110 override fun notifyMessageReadyForRouting(msg: VesMessage) {
111 val now = Instant.now()
112 processingTimeWithoutRouting.record(Duration.between(msg.wtpFrame.receivedAt, now))
113 totalLatencyWithoutRouting.record(Duration.between(epochMicroToInstant(msg.header.lastEpochMicrosec), now))
116 override fun notifyMessageReceived(msg: WireFrameMessage) {
117 receivedMessages.increment()
118 receivedMessagesPayloadBytes.increment(msg.payloadSize.toDouble())
121 override fun notifyMessageReceived(msg: VesMessage) {
122 travelTimeToCollector.record(
123 Duration.between(epochMicroToInstant(msg.header.lastEpochMicrosec), msg.wtpFrame.receivedAt)
127 override fun notifyMessageSent(msg: RoutedMessage) {
128 val now = Instant.now()
129 sentMessages.increment()
130 sentMessagesByTopic(msg.targetTopic).increment()
132 processingTime.record(Duration.between(msg.message.wtpFrame.receivedAt, now))
133 totalLatency.record(Duration.between(epochMicroToInstant(msg.message.header.lastEpochMicrosec), now))
136 override fun notifyMessageDropped(cause: MessageDropCause) {
137 droppedMessages.increment()
138 messagesDroppedByCause(cause.tag).increment()
141 override fun notifyClientRejected(cause: ClientRejectionCause) {
142 clientsRejected.increment()
143 clientsRejectedByCause(cause.tag).increment()
146 override fun notifyClientConnected() {
147 totalConnections.increment()
150 override fun notifyClientDisconnected() {
151 disconnections.increment()
155 val INSTANCE by lazy { MicrometerMetrics() }
156 internal const val PREFIX = "hvves"
157 internal const val MESSAGES = "messages"
158 internal const val RECEIVED = "received"
159 internal const val DISCONNECTIONS = "disconnections"
160 internal const val CONNECTIONS = "connections"
161 internal const val ACTIVE = "active"
162 internal const val BYTES = "bytes"
163 internal const val DATA = "data"
164 internal const val SENT = "sent"
165 internal const val PROCESSING = "processing"
166 internal const val CAUSE = "cause"
167 internal const val CLIENTS = "clients"
168 internal const val REJECTED = "rejected"
169 internal const val TOPIC = "topic"
170 internal const val DROPPED = "dropped"
171 internal const val TIME = "time"
172 internal const val LATENCY = "latency"
173 internal const val PAYLOAD = "payload"
174 internal const val WITHOUT = "without"
175 internal const val ROUTING = "routing"
176 internal const val TRAVEL = "travel"
177 internal const val TO = "to"
178 internal const val COLLECTOR = "collector"
179 internal val MAX_BUCKET_DURATION = Duration.ofSeconds(300L)
180 internal fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"