gerrit.onap Code Review - dcaegen2/collectors/hv-ves.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20 package org.onap.dcae.collectors.veshv.tests.component
  21
  22 import arrow.syntax.function.partially1
  23 import io.netty.buffer.*
  24 import org.assertj.core.api.Assertions.assertThat
  25 import org.jetbrains.spek.api.Spek
  26 import org.jetbrains.spek.api.dsl.describe
  27 import org.jetbrains.spek.api.dsl.it
  28 import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder
  29 import org.onap.dcae.collectors.veshv.tests.fakes.CountingSink
  30 import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
  31 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
  32 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
  33 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.VALID
  34 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain.HVRANMEAS
  35 import reactor.core.publisher.Flux
  36 import reactor.math.sum
  37 import java.security.MessageDigest
  38 import java.time.Duration
  39 import java.util.*
  40 import kotlin.system.measureTimeMillis
  41
  42 /**
  43  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  44  * @since May 2018
  45  */
  46 object PerformanceSpecification : Spek({
  47     debugRx(false)
  48
  49     describe("VES High Volume Collector performance") {
  50         it("should handle multiple clients in reasonable time") {
  51             val sink = CountingSink()
  52             val sut = Sut(sink)
  53             sut.configurationProvider.updateConfiguration(basicConfiguration)
  54
  55             val numMessages: Long = 300_000
  56             val runs = 4
  57             val timeout = Duration.ofMinutes((1 + (runs / 2)).toLong())
  58
  59             val params = MessageParameters(
  60                     domain = HVRANMEAS,
  61                     messageType = VALID,
  62                     amount = numMessages)
  63
  64             val fluxes = (1.rangeTo(runs)).map {
  65                 sut.collector.handleConnection(sut.alloc, generateDataStream(sut.alloc, params))
  66             }
  67             val durationMs = measureTimeMillis {
  68                 Flux.merge(fluxes).then().block(timeout)
  69             }
  70
  71             val durationSec = durationMs / 1000.0
  72             val throughput = sink.count / durationSec
  73             logger.info("Processed $runs connections each containing $numMessages msgs.")
  74             logger.info("Forwarded ${sink.count / ONE_MILION} Mmsgs in $durationSec seconds, that is $throughput msgs/s")
  75             assertThat(sink.count)
  76                     .describedAs("should send all events")
  77                     .isEqualTo(runs * numMessages)
  78         }
  79
  80         it("should disconnect on transmission errors") {
  81             val sink = CountingSink()
  82             val sut = Sut(sink)
  83             sut.configurationProvider.updateConfiguration(basicConfiguration)
  84
  85             val numMessages: Long = 100_000
  86             val timeout = Duration.ofSeconds(30)
  87
  88             val params = MessageParameters(
  89                     domain = HVRANMEAS,
  90                     messageType = VALID,
  91                     amount = numMessages)
  92
  93             val dataStream = generateDataStream(sut.alloc, params)
  94                     .transform(::dropWhenIndex.partially1 { it % 101 == 0L })
  95             sut.collector.handleConnection(sut.alloc, dataStream)
  96                     .timeout(timeout)
  97                     .block()
  98
  99             logger.info("Forwarded ${sink.count} msgs")
 100             assertThat(sink.count)
 101                     .describedAs("should send up to number of events")
 102                     .isLessThan(numMessages)
 103         }
 104     }
 105
 106     describe("test infrastructure") {
 107         val digest = MessageDigest.getInstance("MD5")
 108
 109         fun collectDigest(bb: ByteBuf) {
 110             bb.markReaderIndex()
 111             while (bb.isReadable) {
 112                 digest.update(bb.readByte())
 113             }
 114             bb.resetReaderIndex()
 115         }
 116
 117         fun calculateDigest(arrays: List<ByteArray>): ByteArray {
 118             for (array in arrays) {
 119                 digest.update(array)
 120             }
 121             return digest.digest()
 122         }
 123
 124         it("should yield same bytes as in the input") {
 125             val numberOfBuffers = 10
 126             val singleBufferSize = 1000
 127             val arrays = (1.rangeTo(numberOfBuffers)).map { randomByteArray(singleBufferSize) }
 128             val inputDigest = calculateDigest(arrays)
 129
 130             val actualTotalSize = Flux.fromIterable(arrays)
 131                     .map { Unpooled.wrappedBuffer(it) }
 132                     .transform { simulateRemoteTcp(UnpooledByteBufAllocator.DEFAULT, 4, it) }
 133                     .doOnNext(::collectDigest)
 134                     .map {
 135                         val size = it.readableBytes()
 136                         it.release()
 137                         size
 138                     }
 139                     .sum()
 140                     .map(Long::toInt)
 141                     .block()
 142
 143             val outputDigest = digest.digest()
 144
 145             assertThat(actualTotalSize).isEqualTo(numberOfBuffers * singleBufferSize)
 146             assertThat(outputDigest).isEqualTo(inputDigest)
 147
 148         }
 149     }
 150 })
 151
 152
 153 private const val ONE_MILION = 1_000_000.0
 154
 155 private val rand = Random()
 156 private fun randomByteArray(size: Int): ByteArray {
 157     val bytes = ByteArray(size)
 158     rand.nextBytes(bytes)
 159     return bytes
 160 }
 161
 162 fun dropWhenIndex(predicate: (Long) -> Boolean, stream: Flux<ByteBuf>): Flux<ByteBuf> =
 163         stream.index()
 164                 .filter { predicate(it.t1) }
 165                 .map { it.t2 }
 166
 167 private fun generateDataStream(alloc: ByteBufAllocator, params: MessageParameters): Flux<ByteBuf> =
 168         WireFrameEncoder(alloc).let { encoder ->
 169             MessageGenerator.INSTANCE
 170                     .createMessageFlux(listOf(params))
 171                     .map(encoder::encode)
 172                     .transform { simulateRemoteTcp(alloc, 1000, it) }
 173         }
 174
 175 private fun simulateRemoteTcp(alloc: ByteBufAllocator, maxSize: Int, byteBuffers: Flux<ByteBuf>) =
 176         byteBuffers
 177                 .bufferTimeout(maxSize, Duration.ofMillis(250))
 178                 .map { joinBuffers(alloc, it) }
 179                 .concatMap { randomlySplitTcpFrames(it) }
 180
 181 private fun joinBuffers(alloc: ByteBufAllocator, it: List<ByteBuf>?) =
 182         alloc.compositeBuffer().addComponents(true, it)
 183
 184 private fun randomlySplitTcpFrames(bb: CompositeByteBuf): Flux<ByteBuf> {
 185     val targetFrameSize = Math.max(4, (bb.readableBytes() * Math.random()).toInt())
 186     return Flux.create<ByteBuf> { sink ->
 187         while (bb.isReadable) {
 188             val frameSize = Math.min(targetFrameSize, bb.readableBytes())
 189             sink.next(bb.retainedSlice(bb.readerIndex(), frameSize))
 190             bb.readerIndex(bb.readerIndex() + frameSize)
 191         }
 192         bb.release()
 193         sink.complete()
 194     }
 195 }