diff options
author | 2018-06-15 16:09:41 +0200 | |
---|---|---|
committer | 2018-08-02 10:27:38 +0200 | |
commit | 553154ae42e5362dacab6c190b8cf1e1388f5b87 (patch) | |
tree | 9f78fdd5da58082883724ead537dd5368c2cf940 /hv-collector-core/src/main | |
parent | 69f43f2532106b9d2d4c64fabbf4ea092e4dbf4a (diff) |
Change Consul configuration update policy
- At startup default config is applied
- Configuration is updated in intervals given at VES-HV
service startup to allow dynamic changes
- Included consul service startup in docker-compose file
- VES-HV now exits when fails to acquire confguration from consul
Closes ONAP-229
Change-Id: I896cfd177fa45381f9822278c2dffc113dd3df72
Signed-off-by: jakub.dudycz@nokia.com
Issue-ID: DCAEGEN2-601
Diffstat (limited to 'hv-collector-core/src/main')
6 files changed, 61 insertions, 43 deletions
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt index 8785180b..06047fd4 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt @@ -49,7 +49,9 @@ class CollectorFactory(val configuration: ConfigurationProvider, } private fun createVesHvCollector(): Flux<Collector> = - configuration().map(this::createVesHvCollector) + configuration() + .doOnError { System.exit(ERROR_CODE) } + .map(this::createVesHvCollector) private fun createVesHvCollector(config: CollectorConfiguration): Collector { return VesHvCollector( @@ -61,5 +63,8 @@ class CollectorFactory(val configuration: ConfigurationProvider, metrics = metrics) } + companion object { + const val ERROR_CODE = 3 + } } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt index 358be108..d9e7432d 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt @@ -22,9 +22,8 @@ package org.onap.dcae.collectors.veshv.impl.adapters import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider import org.onap.dcae.collectors.veshv.boundary.SinkProvider import org.onap.dcae.collectors.veshv.impl.adapters.kafka.KafkaSinkProvider -import org.onap.dcae.collectors.veshv.model.CollectorConfiguration -import reactor.core.publisher.Flux import reactor.ipc.netty.http.client.HttpClient +import java.time.Duration /** * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> @@ -34,13 +33,8 @@ object AdapterFactory { fun kafkaSink(): SinkProvider = KafkaSinkProvider() fun loggingSink(): SinkProvider = LoggingSinkProvider() - fun staticConfigurationProvider(config: CollectorConfiguration) = - object : ConfigurationProvider { - override fun invoke() = Flux.just(config) - } - - fun consulConfigurationProvider(url: String): ConfigurationProvider = - ConsulConfigurationProvider(url, httpAdapter()) + fun consulConfigurationProvider(url: String, updateInterval: Duration): ConfigurationProvider = + ConsulConfigurationProvider(url, updateInterval, httpAdapter()) fun httpAdapter(): HttpAdapter = HttpAdapter(HttpClient.create()) } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt index 04e4927d..c70d128a 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt @@ -21,12 +21,14 @@ package org.onap.dcae.collectors.veshv.impl.adapters import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider import org.onap.dcae.collectors.veshv.model.CollectorConfiguration -import org.onap.ves.VesEventV5 +import org.onap.dcae.collectors.veshv.model.routing +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain.HVRANMEAS +import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain.forNumber import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import java.io.StringReader import java.time.Duration -import java.util.* +import java.util.Base64 import java.util.concurrent.atomic.AtomicReference import javax.json.Json import javax.json.JsonObject @@ -36,20 +38,39 @@ import javax.json.JsonObject * @author Jakub Dudycz <jakub.dudycz@nokia.com> * @since May 2018 */ -internal class ConsulConfigurationProvider(private val url: String, private val http: HttpAdapter) - : ConfigurationProvider { +internal class ConsulConfigurationProvider(private val url: String, + private val updateInterval: Duration, + private val http: HttpAdapter +) : ConfigurationProvider { private val logger = LoggerFactory.getLogger(ConsulConfigurationProvider::class.java) private var lastConfigurationHash: AtomicReference<Int> = AtomicReference() override fun invoke(): Flux<CollectorConfiguration> = - Flux.interval(Duration.ZERO, REFRESH_INTERVAL) - .flatMap { http.getResponse(url) } - .filter { body -> body.hashCode() != lastConfigurationHash.get() } - .doOnNext { body -> lastConfigurationHash.set(body.hashCode()) } - .map { str -> getConfigurationJson(str) } - .map { json -> createCollectorConfiguration(json) } + Flux.concat(createDefaultConfigurationFlux(), createConsulFlux()) + + private fun createDefaultConfigurationFlux(): Flux<CollectorConfiguration> = Flux.just( + CollectorConfiguration( + kafkaBootstrapServers = "kafka:9092", + routing = routing { + defineRoute { + fromDomain(HVRANMEAS) + toTopic("ves_hvRanMeas") + withFixedPartitioning() + } + }.build()) + ).doOnNext { logger.info("Applied default configuration") } + + private fun createConsulFlux(): Flux<CollectorConfiguration> = Flux.interval(updateInterval) + .flatMap { http.get(url) } + .doOnError { logger.error("Encountered an error when trying to acquire configuration from consul. " + + "Shutting down..") } + .filter { it.hashCode() != lastConfigurationHash.get() } + .doOnNext { lastConfigurationHash.set(it.hashCode()) } + .map { getConfigurationJson(it) } + .map { createCollectorConfiguration(it) } + private fun getConfigurationJson(str: String): JsonObject { val response = Json.createReader(StringReader(str)).readArray().getJsonObject(0) @@ -60,23 +81,21 @@ internal class ConsulConfigurationProvider(private val url: String, private val } private fun createCollectorConfiguration(configuration: JsonObject): CollectorConfiguration { - - val routing = configuration.getJsonObject("routing") + val routing = configuration.getJsonArray("routing") return CollectorConfiguration( kafkaBootstrapServers = configuration.getString("kafkaBootstrapServers"), routing = org.onap.dcae.collectors.veshv.model.routing { - defineRoute { - fromDomain(VesEventV5.VesEvent.CommonEventHeader.Domain.forNumber(routing.getInt("fromDomain"))) - toTopic(routing.getString("toTopic")) - withFixedPartitioning() + for (route in routing) { + val routeObj = route.asJsonObject() + defineRoute { + fromDomain(forNumber(routeObj.getInt("fromDomain"))) + toTopic(routeObj.getString("toTopic")) + withFixedPartitioning() + } } }.build() ) } - - companion object { - private const val REFRESH_INTERVAL_MINUTES: Long = 5 - private val REFRESH_INTERVAL = Duration.ofMinutes(REFRESH_INTERVAL_MINUTES) - } } + diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt index 236e3cb7..a41cd09f 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt @@ -34,15 +34,12 @@ open class HttpAdapter(private val httpClient: HttpClient) { private val logger = LoggerFactory.getLogger(HttpAdapter::class.java) - open fun getResponse(url: String): Mono<String> = + open fun get(url: String): Mono<String> = httpClient.get(url) - .onErrorResume { e -> unableToGetResource(e, url) } - .flatMap { res -> res.receiveContent().toMono() } - .map { content -> content.content().toString(Charset.defaultCharset()) } - - - private fun unableToGetResource(e: Throwable, url: String): Mono<HttpClientResponse> { - logger.info("Failed to get resource on path: $url\n${e.localizedMessage}") - return Mono.empty() - } + .doOnError { + logger.error("Failed to get resource on path: $url (${it.localizedMessage})") + logger.debug("Nested exception:", it) + } + .flatMap { it.receiveContent().toMono() } + .map { it.content().toString(Charset.defaultCharset()) } } diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt index 580d36c5..056e0557 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt @@ -31,10 +31,12 @@ import reactor.core.publisher.Flux * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> * @since May 2018 */ -internal class WireChunkDecoder(private val decoder: WireFrameDecoder, alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) { +internal class WireChunkDecoder(private val decoder: WireFrameDecoder, + alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) { private val streamBuffer = alloc.compositeBuffer() - fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter.createFlux(decoder, streamBuffer, byteBuf) + fun decode(byteBuf: ByteBuf): Flux<WireFrame> = StreamBufferEmitter + .createFlux(decoder, streamBuffer, byteBuf) .doOnSubscribe { logIncomingMessage(byteBuf) } .doOnNext(this::logDecodedWireMessage) diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServerConfiguration.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServerConfiguration.kt index 67a7d6f2..025c59f6 100644 --- a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServerConfiguration.kt +++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServerConfiguration.kt @@ -29,6 +29,7 @@ import java.time.Duration data class ServerConfiguration( val port: Int, val configurationUrl: String, + val configurationUpdateInterval: Duration, val securityConfiguration: SecurityConfiguration, val idleTimeout: Duration, val dummyMode: Boolean = false) |