1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
/*
* ============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.tests.component
import com.google.protobuf.ByteString
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.given
import org.jetbrains.spek.api.dsl.it
import org.jetbrains.spek.api.dsl.on
import org.onap.dcae.collectors.veshv.domain.VesEventDomain
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP
import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
import org.onap.dcae.collectors.veshv.model.MessageDropCause.INVALID_MESSAGE
import org.onap.dcae.collectors.veshv.model.MessageDropCause.KAFKA_FAILURE
import org.onap.dcae.collectors.veshv.model.MessageDropCause.ROUTE_NOT_FOUND
import org.onap.dcae.collectors.veshv.tests.fakes.MEASUREMENTS_FOR_VF_SCALING_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.PERF3GPP_TOPIC
import org.onap.dcae.collectors.veshv.tests.fakes.configWithBasicRouting
import org.onap.dcae.collectors.veshv.tests.fakes.configWithTwoDomainsToOneTopicRouting
import org.onap.dcae.collectors.veshv.tests.utils.garbageFrame
import org.onap.dcae.collectors.veshv.tests.utils.messageWithInvalidListenerVersion
import org.onap.dcae.collectors.veshv.tests.utils.messageWithInvalidWireFrameHeader
import org.onap.dcae.collectors.veshv.tests.utils.messageWithPayloadOfSize
import org.onap.dcae.collectors.veshv.tests.utils.vesEvent
import org.onap.dcae.collectors.veshv.tests.utils.vesWireFrameMessage
import org.onap.dcae.collectors.veshv.tests.utils.wireFrameMessageWithInvalidPayload
import java.time.Duration
object MetricsSpecification : Spek({
debugRx(false)
describe("Bytes received metrics") {
it("should sum up all bytes received") {
val sut = vesHvWithAlwaysSuccessfulSink()
val vesWireFrameMessage = vesWireFrameMessage()
val invalidWireFrame = messageWithInvalidWireFrameHeader()
val bytesSent = invalidWireFrame.readableBytes() +
vesWireFrameMessage.readableBytes()
sut.handleConnection(
vesWireFrameMessage,
invalidWireFrame
)
val metrics = sut.metrics
assertThat(metrics.bytesReceived)
.describedAs("bytesReceived metric")
.isEqualTo(bytesSent)
}
}
describe("Messages received metrics") {
it("should sum up all received messages bytes") {
val sut = vesHvWithAlwaysSuccessfulSink()
val firstVesEvent = vesEvent(eventFields = ByteString.copyFrom(ByteArray(10)))
val secondVesEvent = vesEvent(eventFields = ByteString.copyFrom(ByteArray(40)))
val firstVesMessage = vesWireFrameMessage(firstVesEvent)
val secondVesMessage = vesWireFrameMessage(secondVesEvent)
val serializedMessagesSize = firstVesEvent.serializedSize + secondVesEvent.serializedSize
sut.handleConnection(
firstVesMessage,
secondVesMessage
)
val metrics = sut.metrics
assertThat(metrics.messageBytesReceived)
.describedAs("messageBytesReceived metric")
.isEqualTo(serializedMessagesSize)
}
}
describe("Messages sent metrics") {
it("should gather info for each topic separately") {
val sut = vesHvWithAlwaysSuccessfulSink(configWithTwoDomainsToOneTopicRouting)
sut.handleConnection(
vesWireFrameMessage(PERF3GPP),
vesWireFrameMessage(PERF3GPP),
vesWireFrameMessage(VesEventDomain.MEASUREMENT)
)
val metrics = sut.metrics
assertThat(metrics.messagesSentCount)
.describedAs("messagesSentCount metric")
.isEqualTo(3)
assertThat(metrics.messagesOnTopic(PERF3GPP_TOPIC))
.describedAs("messagesSentToTopic $PERF3GPP_TOPIC metric")
.isEqualTo(2)
assertThat(metrics.messagesOnTopic(MEASUREMENTS_FOR_VF_SCALING_TOPIC))
.describedAs("messagesSentToTopic $MEASUREMENTS_FOR_VF_SCALING_TOPIC metric")
.isEqualTo(1)
}
}
describe("Processing time") {
it("should gather processing time metric") {
val delay = Duration.ofMillis(10)
val sut = vesHvWithDelayingSink(delay)
sut.handleConnection(vesWireFrameMessage(PERF3GPP))
val metrics = sut.metrics
assertThat(metrics.lastProcessingTimeMicros)
.describedAs("processingTime metric")
.isGreaterThanOrEqualTo(delay.toNanos().toDouble() / 1000.0)
}
}
describe("Messages dropped metrics") {
it("should gather metrics for invalid messages") {
val sut = vesHvWithAlwaysSuccessfulSink(configWithBasicRouting)
sut.handleConnection(
messageWithInvalidWireFrameHeader(),
wireFrameMessageWithInvalidPayload(),
vesWireFrameMessage(domain = PERF3GPP),
messageWithInvalidListenerVersion()
)
val metrics = sut.metrics
assertThat(metrics.messagesDropped(INVALID_MESSAGE))
.describedAs("messagesDroppedCause $INVALID_MESSAGE metric")
.isEqualTo(3)
}
it("should gather metrics for route not found") {
val sut = vesHvWithAlwaysSuccessfulSink(configWithBasicRouting)
sut.handleConnection(
vesWireFrameMessage(domain = PERF3GPP),
vesWireFrameMessage(domain = HEARTBEAT)
)
val metrics = sut.metrics
assertThat(metrics.messagesDropped(ROUTE_NOT_FOUND))
.describedAs("messagesDroppedCause $ROUTE_NOT_FOUND metric")
.isEqualTo(1)
}
it("should gather metrics for sing errors") {
val sut = vesHvWithAlwaysFailingSink(configWithBasicRouting)
sut.handleConnection(vesWireFrameMessage(domain = PERF3GPP))
val metrics = sut.metrics
assertThat(metrics.messagesDropped(KAFKA_FAILURE))
.describedAs("messagesDroppedCause $KAFKA_FAILURE metric")
.isEqualTo(1)
}
it("should gather summed metrics for dropped messages") {
val sut = vesHvWithAlwaysSuccessfulSink(configWithBasicRouting)
sut.handleConnection(
vesWireFrameMessage(domain = PERF3GPP),
vesWireFrameMessage(domain = HEARTBEAT),
wireFrameMessageWithInvalidPayload()
)
val metrics = sut.metrics
assertThat(metrics.messagesDroppedCount)
.describedAs("messagesDroppedCount metric")
.isEqualTo(2)
}
}
describe("clients rejected metrics") {
given("rejection causes") {
mapOf(
ClientRejectionCause.PAYLOAD_SIZE_EXCEEDED_IN_MESSAGE to
messageWithPayloadOfSize(Sut.MAX_PAYLOAD_SIZE_BYTES + 1),
ClientRejectionCause.INVALID_WIRE_FRAME_MARKER to garbageFrame()
).forEach { cause, vesMessage ->
on("cause $cause") {
it("should notify correct metrics") {
val sut = vesHvWithAlwaysSuccessfulSink()
sut.handleConnection(vesMessage)
val metrics = sut.metrics
assertThat(metrics.clientRejectionCause.size)
.describedAs("metrics were notified with only one rejection cause")
.isOne()
assertThat(metrics.clientRejectionCause[cause])
.describedAs("metrics were notified only once with correct client rejection cause")
.isOne()
}
}
}
}
}
})
|
