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 totalLatency = Timer.builder(name(MESSAGES, LATENCY))
70 .maximumExpectedValue(MAX_BUCKET_DURATION)
71 .publishPercentileHistogram(true)
74 private val sentMessages = registry.counter(name(MESSAGES, SENT))
75 private val sentMessagesByTopic = { topic: String ->
76 registry.counter(name(MESSAGES, SENT, TOPIC), TOPIC, topic)
77 }.memoize<String, Counter>()
78 private val droppedMessages = registry.counter(name(MESSAGES, DROPPED))
79 private val messagesDroppedByCause = { cause: String ->
80 registry.counter(name(MESSAGES, DROPPED, CAUSE), CAUSE, cause)
81 }.memoize<String, Counter>()
82 private val clientsRejected = registry.counter(name(CLIENTS, REJECTED))
83 private val clientsRejectedByCause = { cause: String ->
84 registry.counter(name(CLIENTS, REJECTED, CAUSE), CAUSE, cause)
85 }.memoize<String, Counter>()
89 registry.gauge(name(CONNECTIONS, ACTIVE), this) {
90 (totalConnections.count() - disconnections.count()).coerceAtLeast(0.0)
93 ClassLoaderMetrics().bindTo(registry)
94 JvmMemoryMetrics().bindTo(registry)
95 JvmGcMetrics().bindTo(registry)
96 ProcessorMetrics().bindTo(registry)
97 JvmThreadMetrics().bindTo(registry)
100 val metricsProvider = MicrometerPrometheusMetricsProvider(registry)
102 override fun notifyBytesReceived(size: Int) {
103 receivedBytes.increment(size.toDouble())
106 override fun notifyMessageReadyForRouting(msg: VesMessage) {
107 processingTimeWithoutRouting.record(Duration.between(msg.wtpFrame.receivedAt, Instant.now()))
110 override fun notifyMessageReceived(msg: WireFrameMessage) {
111 receivedMessages.increment()
112 receivedMessagesPayloadBytes.increment(msg.payloadSize.toDouble())
115 override fun notifyMessageReceived(msg: VesMessage) {
116 travelTimeToCollector.record(
117 Duration.between(epochMicroToInstant(msg.header.lastEpochMicrosec), msg.wtpFrame.receivedAt)
121 override fun notifyMessageSent(msg: RoutedMessage) {
122 val now = Instant.now()
123 sentMessages.increment()
124 sentMessagesByTopic(msg.targetTopic).increment()
126 processingTime.record(Duration.between(msg.message.wtpFrame.receivedAt, now))
127 totalLatency.record(Duration.between(epochMicroToInstant(msg.message.header.lastEpochMicrosec), now))
130 override fun notifyMessageDropped(cause: MessageDropCause) {
131 droppedMessages.increment()
132 messagesDroppedByCause(cause.tag).increment()
135 override fun notifyClientRejected(cause: ClientRejectionCause) {
136 clientsRejected.increment()
137 clientsRejectedByCause(cause.tag).increment()
140 override fun notifyClientConnected() {
141 totalConnections.increment()
144 override fun notifyClientDisconnected() {
145 disconnections.increment()
149 val INSTANCE by lazy { MicrometerMetrics() }
150 internal const val PREFIX = "hvves"
151 internal const val MESSAGES = "messages"
152 internal const val RECEIVED = "received"
153 internal const val DISCONNECTIONS = "disconnections"
154 internal const val CONNECTIONS = "connections"
155 internal const val ACTIVE = "active"
156 internal const val BYTES = "bytes"
157 internal const val DATA = "data"
158 internal const val SENT = "sent"
159 internal const val PROCESSING = "processing"
160 internal const val CAUSE = "cause"
161 internal const val CLIENTS = "clients"
162 internal const val REJECTED = "rejected"
163 internal const val TOPIC = "topic"
164 internal const val DROPPED = "dropped"
165 internal const val TIME = "time"
166 internal const val LATENCY = "latency"
167 internal const val PAYLOAD = "payload"
168 internal const val WITHOUT = "without"
169 internal const val ROUTING = "routing"
170 internal const val TRAVEL = "travel"
171 internal const val TO = "to"
172 internal const val COLLECTOR = "collector"
173 internal val MAX_BUCKET_DURATION = Duration.ofSeconds(300L)
174 internal fun name(vararg name: String) = "$PREFIX.${name.joinToString(".")}"