<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
<artifactId>cbs-client</artifactId>
</dependency>
-
+ <dependency>
+ <groupId>io.projectreactor.addons</groupId>
+ <artifactId>reactor-extra</artifactId>
+ </dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jdk8</artifactId>
import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.MissingArgumentException
import org.onap.dcae.collectors.veshv.config.api.model.ValidationException
-import org.onap.dcae.collectors.veshv.config.impl.HvVesCommandLineParser
+import org.onap.dcae.collectors.veshv.config.impl.CbsConfigurationProvider
+import org.onap.dcae.collectors.veshv.config.impl.ConfigurationMerger
import org.onap.dcae.collectors.veshv.config.impl.ConfigurationValidator
import org.onap.dcae.collectors.veshv.config.impl.FileConfigurationReader
+import org.onap.dcae.collectors.veshv.config.impl.HvVesCommandLineParser
+import org.onap.dcae.collectors.veshv.utils.arrow.rightOrThrow
import org.onap.dcae.collectors.veshv.utils.arrow.throwOnLeft
+import org.onap.dcae.collectors.veshv.utils.logging.MappedDiagnosticContext
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties
import reactor.core.publisher.Flux
class ConfigurationModule {
private val configReader = FileConfigurationReader()
private val configValidator = ConfigurationValidator()
- private lateinit var initialConfig: HvVesConfiguration
-
fun healthCheckPort(args: Array<String>): Int = cmd.getHealthcheckPort(args)
- fun hvVesConfigurationUpdates(args: Array<String>): Flux<HvVesConfiguration> =
+ fun hvVesConfigurationUpdates(args: Array<String>,
+ configStateListener: ConfigurationStateListener,
+ mdc: MappedDiagnosticContext): Flux<HvVesConfiguration> =
Flux.just(cmd.getConfigurationFile(args))
.throwOnLeft { MissingArgumentException(it.message, it.cause) }
.map { it.reader().use(configReader::loadConfig) }
- .map { configValidator.validate(it) }
- .throwOnLeft { ValidationException(it.message) }
- .doOnNext { initialConfig = it }
+ .cache()
+ .flatMap { basePartialConfig ->
+ val baseConfig = configValidator.validate(basePartialConfig)
+ .rightOrThrow { ValidationException(it.message) }
+ val cbsConfigProvider = CbsConfigurationProvider(
+ CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
+ baseConfig.cbs,
+ configStateListener,
+ mdc)
+ val merger = ConfigurationMerger()
+ cbsConfigProvider()
+ .map { merger.merge(basePartialConfig, it) }
+ .map { configValidator.validate(it) }
+ .throwOnLeft { ValidationException(it.message) }
+ }
}
* ============LICENSE_START=======================================================
* dcaegen2-collectors-veshv
* ================================================================================
- * Copyright (C) 2018 NOKIA
+ * Copyright (C) 2019 NOKIA
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl.adapters
+package org.onap.dcae.collectors.veshv.config.api
-class ParsingException(message: String, cause: Throwable) : Exception(message, cause)
+interface ConfigurationStateListener {
+ fun retrying() {}
+}
)
data class CollectorConfiguration(
- val maxRequestSizeBytes: Int,
- val kafkaServers: String,
val routing: Routing
)
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl.adapters
+package org.onap.dcae.collectors.veshv.config.impl
+import arrow.core.None
+import arrow.core.Option
+import arrow.core.Some
import com.google.gson.JsonObject
-import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.Route
import org.onap.dcae.collectors.veshv.config.api.model.Routing
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
-import org.onap.dcae.collectors.veshv.model.ServiceContext
import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.dcae.collectors.veshv.utils.logging.MappedDiagnosticContext
import org.onap.dcae.collectors.veshv.utils.logging.onErrorLog
import org.onap.dcaegen2.services.sdk.model.streams.StreamType.KAFKA
import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
import reactor.core.publisher.Mono
import reactor.retry.Jitter
import reactor.retry.Retry
-import java.time.Duration
/**
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since May 2018
*/
-internal class ConfigurationProviderImpl(private val cbsClientMono: Mono<CbsClient>,
- private val firstRequestDelay: Duration,
- private val requestInterval: Duration,
- private val healthState: HealthState,
- private val streamParser: StreamFromGsonParser<KafkaSink>,
- retrySpec: Retry<Any>
+internal class CbsConfigurationProvider(private val cbsClientMono: Mono<CbsClient>,
+ private val cbsConfiguration: CbsConfiguration,
+ private val streamParser: StreamFromGsonParser<KafkaSink>,
+ private val configurationStateListener: ConfigurationStateListener,
+ retrySpec: Retry<Any>,
+ private val mdc: MappedDiagnosticContext
-) : ConfigurationProvider {
- constructor(cbsClientMono: Mono<CbsClient>, params: CbsConfiguration) : this(
- cbsClientMono,
- params.firstRequestDelay,
- params.requestInterval,
- HealthState.INSTANCE,
- StreamFromGsonParsers.kafkaSinkParser(),
- Retry.any<Any>()
- .retryMax(MAX_RETRIES)
- .fixedBackoff(params.requestInterval)
- .jitter(Jitter.random())
- )
+) {
+ constructor(cbsClientMono: Mono<CbsClient>,
+ cbsConfig: CbsConfiguration,
+ configurationStateListener: ConfigurationStateListener,
+ mdc: MappedDiagnosticContext) :
+ this(
+ cbsClientMono,
+ cbsConfig,
+ StreamFromGsonParsers.kafkaSinkParser(),
+ configurationStateListener,
+ Retry.any<Any>()
+ .retryMax(MAX_RETRIES)
+ .fixedBackoff(cbsConfig.requestInterval)
+ .jitter(Jitter.random()),
+ mdc
+ )
private val retry = retrySpec.doOnRetry {
- logger.withWarn(ServiceContext::mdc) {
+ logger.withWarn(mdc) {
log("Exception from configuration provider client, retrying subscription", it.exception())
}
- healthState.changeState(HealthDescription.RETRYING_FOR_DYNAMIC_CONFIGURATION)
+ configurationStateListener.retrying()
}
- override fun invoke(): Flux<Routing> =
+ operator fun invoke(): Flux<PartialConfiguration> =
cbsClientMono
- .doOnNext { logger.info(ServiceContext::mdc) { "CBS client successfully created" } }
- .onErrorLog(logger, ServiceContext::mdc) { "Failed to retrieve CBS client" }
+ .doOnNext { logger.info(mdc) { "CBS client successfully created" } }
+ .onErrorLog(logger, mdc) { "Failed to retrieve CBS client" }
.retryWhen(retry)
- .doFinally { logger.trace(ServiceContext::mdc) { "CBS client subscription finished" } }
+ .doFinally { logger.trace(mdc) { "CBS client subscription finished" } }
.flatMapMany(::handleUpdates)
private fun handleUpdates(cbsClient: CbsClient) = cbsClient
.updates(CbsRequests.getConfiguration(RequestDiagnosticContext.create()),
- firstRequestDelay,
- requestInterval)
- .doOnNext { logger.info(ServiceContext::mdc) { "Received new configuration:\n$it" } }
+ cbsConfiguration.firstRequestDelay,
+ cbsConfiguration.requestInterval)
+ .doOnNext { logger.info(mdc) { "Received new configuration:\n$it" } }
.map(::createRoutingDescription)
- .onErrorLog(logger, ServiceContext::mdc) { "Error while creating configuration" }
+ .onErrorLog(logger, mdc) { "Error while creating configuration" }
.retryWhen(retry)
+ .map { PartialConfiguration(collector = Some(PartialCollectorConfig(routing = it))) }
- private fun createRoutingDescription(configuration: JsonObject): Routing = try {
- DataStreams.namedSinks(configuration)
+ private fun createRoutingDescription(configuration: JsonObject): Option<Routing> = try {
+ val routes = DataStreams.namedSinks(configuration)
.filter(streamOfType(KAFKA))
.map(streamParser::unsafeParse)
.map { Route(it.name(), it) }
.asIterable()
.toList()
+ Some(routes)
} catch (e: NullPointerException) {
- throw ParsingException("Failed to parse configuration", e)
+ logger.withWarn(mdc) {
+ log("Invalid streams configuration", e)
+ }
+ None
}
companion object {
private const val MAX_RETRIES = 5L
- private val logger = Logger(ConfigurationProviderImpl::class)
+ private val logger = Logger(CbsConfigurationProvider::class)
}
}
securityConfiguration,
// TOD0: swap when ConfigurationMerger is implemented
// collectorConfiguration
- CollectorConfiguration(-1,
- "I do not exist. I'm not even a URL :o",
- emptyList()),
+ CollectorConfiguration(emptyList()),
// end TOD0
logLevel
)
*/
package org.onap.dcae.collectors.veshv.config.impl
-import arrow.core.*
+import arrow.core.Either
+import arrow.core.Option
+import arrow.core.Try
+import arrow.core.getOrElse
import org.apache.commons.cli.CommandLine
import org.apache.commons.cli.CommandLineParser
import org.apache.commons.cli.DefaultParser
internal data class PartialCollectorConfig(
val maxRequestSizeBytes: Option<Int> = None,
- val kafkaServers: Option<List<InetSocketAddress>> = None,
+ val kafkaServers: Option<List<InetSocketAddress>> = None, // TOD0: remove properties and simplify this part
val routing: Option<Routing> = None
)
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl.adapters
+package org.onap.dcae.collectors.veshv.config.impl
import com.google.gson.JsonParser
import com.nhaarman.mockitokotlin2.any
import com.nhaarman.mockitokotlin2.eq
import com.nhaarman.mockitokotlin2.mock
+import com.nhaarman.mockitokotlin2.times
+import com.nhaarman.mockitokotlin2.verify
import com.nhaarman.mockitokotlin2.whenever
import org.assertj.core.api.Assertions.assertThat
import org.jetbrains.spek.api.Spek
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.healthcheck.api.HealthDescription
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
+import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
+import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
import org.onap.dcaegen2.services.sdk.model.streams.ImmutableAafCredentials
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers
import reactor.core.publisher.Flux
-
import reactor.core.publisher.Mono
import reactor.retry.Retry
import reactor.test.StepVerifier
* @author Jakub Dudycz <jakub.dudycz@nokia.com>
* @since May 2018
*/
-internal object ConfigurationProviderImplTest : Spek({
+internal object CbsConfigurationProviderTest : Spek({
describe("Configuration provider") {
val cbsClient: CbsClient = mock()
val cbsClientMock: Mono<CbsClient> = Mono.just(cbsClient)
- val healthStateProvider = HealthState.INSTANCE
+ val configStateListener: ConfigurationStateListener = mock()
given("configuration is never in cbs") {
- val configProvider = constructConfigurationProvider(cbsClientMock, healthStateProvider)
+ val configProvider = constructConfigurationProvider(cbsClientMock, configStateListener)
on("waiting for configuration") {
val waitTime = Duration.ofMillis(100)
}
given("valid configuration from cbs") {
- val configProvider = constructConfigurationProvider(cbsClientMock, healthStateProvider)
+ val configProvider = constructConfigurationProvider(cbsClientMock, configStateListener)
on("new configuration") {
whenever(cbsClient.updates(any(), eq(firstRequestDelay), eq(requestInterval)))
StepVerifier.create(configProvider().take(1))
.consumeNextWith {
- val route1 = it.elementAt(0)
- val route2 = it.elementAt(1)
+ val routes = it.collector.orNull()!!.routing.orNull()!!
+ val route1 = routes.elementAt(0)
+ val route2 = routes.elementAt(1)
val receivedSink1 = route1.sink
val receivedSink2 = route2.sink
given("invalid configuration from cbs") {
val iterationCount = 3L
val configProvider = constructConfigurationProvider(
- cbsClientMock, healthStateProvider, iterationCount
+ cbsClientMock, configStateListener, iterationCount
)
on("new configuration") {
.verifyError()
}
- it("should update the health state") {
- StepVerifier.create(healthStateProvider().take(iterationCount))
- .expectNextCount(iterationCount - 1)
- .expectNext(HealthDescription.RETRYING_FOR_DYNAMIC_CONFIGURATION)
- .verifyComplete()
+ it("should call state listener when retrying") {
+ verify(configStateListener, times(iterationCount.toInt())).retrying()
}
}
}
private val streamParser = StreamFromGsonParsers.kafkaSinkParser()
private fun constructConfigurationProvider(cbsClientMono: Mono<CbsClient>,
- healthState: HealthState,
+ configurationStateListener: ConfigurationStateListener,
iterationCount: Long = 1
-): ConfigurationProviderImpl {
+): CbsConfigurationProvider {
val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
- return ConfigurationProviderImpl(
+ return CbsConfigurationProvider(
cbsClientMono,
- firstRequestDelay,
- requestInterval,
- healthState,
+ CbsConfiguration(firstRequestDelay, requestInterval),
streamParser,
- retry
+ configurationStateListener,
+ retry,
+ { mapOf("k" to "v") }
)
}
*/
package org.onap.dcae.collectors.veshv.boundary
-import org.onap.dcae.collectors.veshv.config.api.model.Routing
import org.onap.dcae.collectors.veshv.domain.RoutedMessage
import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
import org.onap.dcae.collectors.veshv.model.ClientContext
operator fun invoke(stream: SinkStream, ctx: ClientContext): Lazy<Sink>
}
-typealias ConfigurationProvider = () -> Flux<Routing>
-
interface Metrics {
fun notifyBytesReceived(size: Int)
fun notifyMessageReceived(msg: WireFrameMessage)
*/
package org.onap.dcae.collectors.veshv.boundary
-import arrow.core.Option
import arrow.effects.IO
import io.netty.buffer.ByteBuf
import org.onap.dcae.collectors.veshv.model.ClientContext
}
interface CollectorProvider : Closeable {
- operator fun invoke(ctx: ClientContext): Option<Collector>
+ operator fun invoke(ctx: ClientContext): Collector
}
interface Server {
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl.adapters
+package org.onap.dcae.collectors.veshv.factory
-import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
-import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
import org.onap.dcae.collectors.veshv.impl.adapters.kafka.KafkaSinkProvider
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
*/
object AdapterFactory {
fun sinkCreatorFactory(): SinkProvider = KafkaSinkProvider()
-
- fun configurationProvider(config: CbsConfiguration): ConfigurationProvider =
- ConfigurationProviderImpl(
- CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
- config)
}
*/
package org.onap.dcae.collectors.veshv.factory
-import arrow.core.Option
import org.onap.dcae.collectors.veshv.boundary.Collector
import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
-import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
import org.onap.dcae.collectors.veshv.boundary.Metrics
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
-import org.onap.dcae.collectors.veshv.config.api.model.Routing
+import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
import org.onap.dcae.collectors.veshv.impl.Router
import org.onap.dcae.collectors.veshv.impl.VesDecoder
import org.onap.dcae.collectors.veshv.impl.VesHvCollector
import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder
import org.onap.dcae.collectors.veshv.model.ClientContext
-import org.onap.dcae.collectors.veshv.model.ServiceContext
-import org.onap.dcae.collectors.veshv.utils.arrow.getOption
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import java.util.concurrent.atomic.AtomicReference
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-class CollectorFactory(private val configuration: ConfigurationProvider,
+class CollectorFactory(private val configuration: CollectorConfiguration,
private val sinkProvider: SinkProvider,
private val metrics: Metrics,
- private val maxPayloadSizeBytes: Int,
- private val healthState: HealthState = HealthState.INSTANCE) {
+ private val maxPayloadSizeBytes: Int) {
fun createVesHvCollectorProvider(): CollectorProvider {
- val config = AtomicReference<Routing>()
- configuration()
- .doOnNext {
- logger.info(ServiceContext::mdc) { "Using updated configuration for new connections" }
- healthState.changeState(HealthDescription.HEALTHY)
- }
- .doOnError {
- logger.error(ServiceContext::mdc) { "Failed to acquire configuration ${it.message}" }
- logger.debug(ServiceContext::mdc) { "Detailed stack trace: $it" }
- healthState.changeState(HealthDescription.DYNAMIC_CONFIGURATION_NOT_FOUND)
- }
- .subscribe(config::set)
return object : CollectorProvider {
- override fun invoke(ctx: ClientContext): Option<Collector> =
- config.getOption().map { createVesHvCollector(it, ctx) }
+ override fun invoke(ctx: ClientContext): Collector =
+ createVesHvCollector(ctx)
override fun close() = sinkProvider.close()
}
}
- private fun createVesHvCollector(routing: Routing, ctx: ClientContext): Collector =
+ private fun createVesHvCollector(ctx: ClientContext): Collector =
VesHvCollector(
clientContext = ctx,
wireChunkDecoder = WireChunkDecoder(WireFrameDecoder(maxPayloadSizeBytes), ctx),
protobufDecoder = VesDecoder(),
- router = Router(routing, sinkProvider, ctx, metrics),
+ router = Router(configuration.routing, sinkProvider, ctx, metrics),
metrics = metrics)
companion object {
private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> {
metrics.notifyClientConnected()
logger.info(clientContext::fullMdc) { "Handling new client connection" }
- return collectorProvider(clientContext).fold(
- {
- logger.warn(clientContext::fullMdc) { "Collector is not ready. Closing connection" }
- nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
- },
- handleClient(clientContext, nettyInbound)
- )
+ val collector = collectorProvider(clientContext)
+ return collector.handleClient(clientContext, nettyInbound)
}
- private fun handleClient(clientContext: ClientContext,
- nettyInbound: NettyInbound): (Collector) -> Mono<Void> = { collector ->
+ private fun Collector.handleClient(clientContext: ClientContext,
+ nettyInbound: NettyInbound) =
withConnectionFrom(nettyInbound) { connection ->
connection
.configureIdleTimeout(clientContext, serverConfiguration.idleTimeout)
.logConnectionClosed(clientContext)
}.run {
- collector.handleConnection(nettyInbound.createDataStream())
+ handleConnection(nettyInbound.createDataStream())
}
- }
private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
onReadIdle(timeout.toMillis()) {
import org.jetbrains.spek.api.Spek
import org.jetbrains.spek.api.dsl.describe
import org.jetbrains.spek.api.dsl.it
+import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP
import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder
import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
describe("VES High Volume Collector performance") {
it("should handle multiple clients in reasonable time") {
val sink = CountingSink()
- val sut = Sut(sink)
- sut.configurationProvider.updateConfiguration(basicRouting)
+ val sut = Sut(CollectorConfiguration(basicRouting), sink)
val numMessages: Long = 300_000
val runs = 4
it("should disconnect on transmission errors") {
val sink = CountingSink()
- val sut = Sut(sink)
- sut.configurationProvider.updateConfiguration(basicRouting)
+ val sut = Sut(CollectorConfiguration(basicRouting), sink)
val numMessages: Long = 100_000
val timeout = Duration.ofSeconds(30)
*/
package org.onap.dcae.collectors.veshv.tests.component
-import arrow.core.getOrElse
import arrow.effects.IO
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufAllocator
import org.onap.dcae.collectors.veshv.boundary.Collector
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.config.api.model.Routing
import org.onap.dcae.collectors.veshv.domain.RoutedMessage
import org.onap.dcae.collectors.veshv.factory.CollectorFactory
import org.onap.dcae.collectors.veshv.tests.fakes.AlwaysFailingSink
import org.onap.dcae.collectors.veshv.tests.fakes.AlwaysSuccessfulSink
import org.onap.dcae.collectors.veshv.tests.fakes.DelayingSink
-import org.onap.dcae.collectors.veshv.tests.fakes.FakeConfigurationProvider
-import org.onap.dcae.collectors.veshv.tests.fakes.FakeHealthState
import org.onap.dcae.collectors.veshv.tests.fakes.FakeMetrics
import org.onap.dcae.collectors.veshv.tests.fakes.StoringSink
import org.onap.dcae.collectors.veshv.tests.fakes.basicRouting
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-class Sut(sink: Sink = StoringSink()) : Closeable {
- val configurationProvider = FakeConfigurationProvider()
- val healthStateProvider = FakeHealthState()
+class Sut(configuration: CollectorConfiguration, sink: Sink = StoringSink()) : Closeable {
val alloc: ByteBufAllocator = UnpooledByteBufAllocator.DEFAULT
val metrics = FakeMetrics()
val sinkProvider = DummySinkProvider(sink)
private val collectorFactory = CollectorFactory(
- configurationProvider,
+ configuration,
sinkProvider,
metrics,
- MAX_PAYLOAD_SIZE_BYTES,
- healthStateProvider
+ MAX_PAYLOAD_SIZE_BYTES
)
private val collectorProvider = collectorFactory.createVesHvCollectorProvider()
val collector: Collector
- get() = collectorProvider(ClientContext(alloc)).getOrElse {
- throw IllegalStateException("Collector not available.")
- }
+ get() = collectorProvider(ClientContext(alloc))
fun handleConnection(sink: StoringSink, vararg packets: ByteBuf): List<RoutedMessage> {
private val timeout = Duration.ofSeconds(10)
fun vesHvWithAlwaysSuccessfulSink(routing: Routing = basicRouting): Sut =
- Sut(AlwaysSuccessfulSink()).apply {
- configurationProvider.updateConfiguration(routing)
- }
+ Sut(CollectorConfiguration(routing), AlwaysSuccessfulSink())
fun vesHvWithAlwaysFailingSink(routing: Routing = basicRouting): Sut =
- Sut(AlwaysFailingSink()).apply {
- configurationProvider.updateConfiguration(routing)
- }
+ Sut(CollectorConfiguration(routing), AlwaysFailingSink())
fun vesHvWithDelayingSink(delay: Duration, routing: Routing = basicRouting): Sut =
- Sut(DelayingSink(delay)).apply {
- configurationProvider.updateConfiguration(routing)
- }
+ Sut(CollectorConfiguration(routing), DelayingSink(delay))
import org.jetbrains.spek.api.dsl.describe
import org.jetbrains.spek.api.dsl.given
import org.jetbrains.spek.api.dsl.it
+import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.config.api.model.Routing
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.MEASUREMENT
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.OTHER
}
it("should be able to direct 2 messages from different domains to one topic") {
- val (sut, sink) = vesHvWithStoringSink()
-
- sut.configurationProvider.updateConfiguration(twoDomainsToOneTopicRouting)
+ val (sut, sink) = vesHvWithStoringSink(twoDomainsToOneTopicRouting)
val messages = sut.handleConnection(sink,
vesWireFrameMessage(PERF3GPP),
}
}
- describe("configuration update") {
-
- val defaultTimeout = Duration.ofSeconds(10)
-
- given("successful configuration change") {
-
- lateinit var sut: Sut
- lateinit var sink: StoringSink
-
- beforeEachTest {
- vesHvWithStoringSink().run {
- sut = first
- sink = second
- }
- }
-
- it("should update collector") {
- val firstCollector = sut.collector
-
- sut.configurationProvider.updateConfiguration(alternativeRouting)
- val collectorAfterUpdate = sut.collector
-
- assertThat(collectorAfterUpdate).isNotSameAs(firstCollector)
- }
-
- it("should start routing messages") {
-
- sut.configurationProvider.updateConfiguration(emptyRouting)
-
- val messages = sut.handleConnection(sink, vesWireFrameMessage(PERF3GPP))
- assertThat(messages).isEmpty()
-
- sut.configurationProvider.updateConfiguration(basicRouting)
-
- val messagesAfterUpdate = sut.handleConnection(sink, vesWireFrameMessage(PERF3GPP))
- assertThat(messagesAfterUpdate).hasSize(1)
- val message = messagesAfterUpdate[0]
-
- assertThat(message.targetTopic).describedAs("routed message topic after configuration'PERF3GPP_REGIONAL change")
- .isEqualTo(PERF3GPP_TOPIC)
- assertThat(message.partition).describedAs("routed message partition")
- .isEqualTo(None)
- }
-
- it("should change domain routing") {
-
- val messages = sut.handleConnection(sink, vesWireFrameMessage(PERF3GPP))
- assertThat(messages).hasSize(1)
- val firstMessage = messages[0]
-
- assertThat(firstMessage.targetTopic).describedAs("routed message topic on initial configuration")
- .isEqualTo(PERF3GPP_TOPIC)
- assertThat(firstMessage.partition).describedAs("routed message partition")
- .isEqualTo(None)
-
-
- sut.configurationProvider.updateConfiguration(alternativeRouting)
-
- val messagesAfterUpdate = sut.handleConnection(sink, vesWireFrameMessage(PERF3GPP))
- assertThat(messagesAfterUpdate).hasSize(2)
- val secondMessage = messagesAfterUpdate[1]
-
- assertThat(secondMessage.targetTopic).describedAs("routed message topic after configuration'PERF3GPP_REGIONAL change")
- .isEqualTo(ALTERNATE_PERF3GPP_TOPIC)
- assertThat(secondMessage.partition).describedAs("routed message partition")
- .isEqualTo(None)
- }
-
- it("should update routing for each client sending one message") {
-
- val messagesAmount = 10
- val messagesForEachTopic = 5
-
- Flux.range(0, messagesAmount).doOnNext {
- if (it == messagesForEachTopic) {
- sut.configurationProvider.updateConfiguration(alternativeRouting)
- }
- }.doOnNext {
- sut.handleConnection(sink, vesWireFrameMessage(PERF3GPP))
- }.then().block(defaultTimeout)
-
-
- val messages = sink.sentMessages
- val firstTopicMessagesCount = messages.count { it.targetTopic == PERF3GPP_TOPIC }
- val secondTopicMessagesCount = messages.count { it.targetTopic == ALTERNATE_PERF3GPP_TOPIC }
-
- assertThat(messages.size).isEqualTo(messagesAmount)
- assertThat(messagesForEachTopic)
- .describedAs("amount of messages routed to each topic")
- .isEqualTo(firstTopicMessagesCount)
- .isEqualTo(secondTopicMessagesCount)
- }
-
- it("should not update routing for client sending continuous stream of messages") {
-
- val messageStreamSize = 10
- val pivot = 5
-
- val incomingMessages = Flux.range(0, messageStreamSize)
- .doOnNext {
- if (it == pivot) {
- sut.configurationProvider.updateConfiguration(alternativeRouting)
- println("config changed")
- }
- }
- .map { vesWireFrameMessage(PERF3GPP) }
-
-
- sut.collector.handleConnection(incomingMessages).block(defaultTimeout)
-
- val messages = sink.sentMessages
- val firstTopicMessagesCount = messages.count { it.targetTopic == PERF3GPP_TOPIC }
- val secondTopicMessagesCount = messages.count { it.targetTopic == ALTERNATE_PERF3GPP_TOPIC }
-
- assertThat(messages.size).isEqualTo(messageStreamSize)
- assertThat(firstTopicMessagesCount)
- .describedAs("amount of messages routed to first topic")
- .isEqualTo(messageStreamSize)
-
- assertThat(secondTopicMessagesCount)
- .describedAs("amount of messages routed to second topic")
- .isEqualTo(0)
- }
-
- it("should mark the application healthy") {
- assertThat(sut.healthStateProvider.currentHealth)
- .describedAs("application health state")
- .isEqualTo(HealthDescription.HEALTHY)
- }
- }
-
- given("failed configuration change") {
- val (sut, _) = vesHvWithStoringSink()
- sut.configurationProvider.shouldThrowExceptionOnConfigUpdate(true)
- sut.configurationProvider.updateConfiguration(basicRouting)
-
- it("should mark the application unhealthy ") {
- assertThat(sut.healthStateProvider.currentHealth)
- .describedAs("application health state")
- .isEqualTo(HealthDescription.DYNAMIC_CONFIGURATION_NOT_FOUND)
- }
- }
- }
-
describe("request validation") {
it("should reject message with payload greater than 1 MiB and all subsequent messages") {
val (sut, sink) = vesHvWithStoringSink()
})
-private fun vesHvWithStoringSink(): Pair<Sut, StoringSink> {
+private fun vesHvWithStoringSink(routing: Routing = basicRouting): Pair<Sut, StoringSink> {
val sink = StoringSink()
- val sut = Sut(sink)
- sut.configurationProvider.updateConfiguration(basicRouting)
+ val sut = Sut(CollectorConfiguration(routing), sink)
return Pair(sut, sink)
}
+++ /dev/null
-/*
- * ============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.fakes
-
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
-import reactor.core.publisher.Flux
-
-class FakeHealthState : HealthState {
-
- lateinit var currentHealth: HealthDescription
-
- override fun changeState(healthDescription: HealthDescription) {
- currentHealth = healthDescription
- }
-
- override fun invoke(): Flux<HealthDescription> {
- throw NotImplementedError()
- }
-}
+++ /dev/null
-/*
- * ============LICENSE_START=======================================================
- * dcaegen2-collectors-veshv
- * ================================================================================
- * Copyright (C) 2018-2019 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.fakes
-
-import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
-import org.onap.dcae.collectors.veshv.config.api.model.Routing
-import reactor.core.publisher.FluxProcessor
-import reactor.core.publisher.UnicastProcessor
-import reactor.retry.RetryExhaustedException
-
-
-class FakeConfigurationProvider : ConfigurationProvider {
- private var shouldThrowException = false
- private val configStream: FluxProcessor<Routing, Routing> = UnicastProcessor.create()
-
- fun updateConfiguration(routing: Routing) =
- if (shouldThrowException) {
- configStream.onError(RetryExhaustedException("I'm so tired"))
- } else {
- configStream.onNext(routing)
- }
-
-
- fun shouldThrowExceptionOnConfigUpdate(shouldThrowException: Boolean) {
- this.shouldThrowException = shouldThrowException
- }
-
- override fun invoke() = configStream
-}
package org.onap.dcae.collectors.veshv.main
import org.onap.dcae.collectors.veshv.config.api.ConfigurationModule
+import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
private val configurationModule = ConfigurationModule()
fun main(args: Array<String>) {
+ val configStateListener = object : ConfigurationStateListener {
+ override fun retrying() {
+ HealthState.INSTANCE.changeState(HealthDescription.RETRYING_FOR_DYNAMIC_CONFIGURATION)
+ }
+ }
+
HealthCheckServer.start(configurationModule.healthCheckPort(args))
configurationModule
- .hvVesConfigurationUpdates(args)
+ .hvVesConfigurationUpdates(args, configStateListener, ServiceContext::mdc)
.publishOn(Schedulers.single(Schedulers.elastic()))
+ .doOnNext {
+ logger.info(ServiceContext::mdc) { "Using updated configuration for new connections" }
+ HealthState.INSTANCE.changeState(HealthDescription.HEALTHY)
+ }
+ .doOnError {
+ logger.error(ServiceContext::mdc) { "Failed to acquire configuration ${it.message}" }
+ logger.debug(ServiceContext::mdc) { "Detailed stack trace: $it" }
+ HealthState.INSTANCE.changeState(HealthDescription.DYNAMIC_CONFIGURATION_NOT_FOUND)
+ }
.doOnNext(::startServer)
.doOnError(::logServerStartFailed)
.neverComplete() // TODO: remove after merging configuration stream with cbs
import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
import org.onap.dcae.collectors.veshv.factory.CollectorFactory
import org.onap.dcae.collectors.veshv.factory.ServerFactory
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
-import org.onap.dcae.collectors.veshv.impl.adapters.AdapterFactory
+import org.onap.dcae.collectors.veshv.factory.AdapterFactory
import org.onap.dcae.collectors.veshv.main.metrics.MicrometerMetrics
import org.onap.dcae.collectors.veshv.model.ServiceContext
import org.onap.dcae.collectors.veshv.utils.ServerHandle
private fun initializeCollectorFactory(config: HvVesConfiguration): CollectorFactory =
CollectorFactory(
- AdapterFactory.configurationProvider(config.cbs),
+ config.collector,
AdapterFactory.sinkCreatorFactory(),
MicrometerMetrics.INSTANCE,
- config.server.maxPayloadSizeBytes,
- HealthState.INSTANCE
+ config.server.maxPayloadSizeBytes
)
private fun logServerStarted(handle: ServerHandle) =
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff