sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt

   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.impl.adapters
  21
  22 import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
  23 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
  24 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
  25 import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
  26 import org.onap.dcae.collectors.veshv.model.ConfigurationProviderParams
  27 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  28 import reactor.core.publisher.Flux
  29 import reactor.core.publisher.Mono
  30 import reactor.retry.Jitter
  31 import reactor.retry.Retry
  32 import java.io.StringReader
  33 import java.time.Duration
  34 import java.util.concurrent.atomic.AtomicReference
  35 import javax.json.Json
  36 import javax.json.JsonObject
  37
  38
  39 /**
  40  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  41  * @since May 2018
  42  */
  43 internal class ConsulConfigurationProvider(private val http: HttpAdapter,
  44                                            private val url: String,
  45                                            private val firstRequestDelay: Duration,
  46                                            private val requestInterval: Duration,
  47                                            private val healthState: HealthState,
  48                                            retrySpec: Retry<Any>
  49
  50 ) : ConfigurationProvider {
  51
  52     private val lastConfigurationHash: AtomicReference<Int> = AtomicReference(0)
  53     private val retry = retrySpec
  54             .doOnRetry {
  55                 logger.warn("Could not get fresh configuration", it.exception())
  56                 healthState.changeState(HealthDescription.RETRYING_FOR_CONSUL_CONFIGURATION)
  57             }
  58
  59     constructor(http: HttpAdapter,
  60                 params: ConfigurationProviderParams) : this(
  61             http,
  62             params.configurationUrl,
  63             params.firstRequestDelay,
  64             params.requestInterval,
  65             HealthState.INSTANCE,
  66             Retry.any<Any>()
  67                     .retryMax(MAX_RETRIES)
  68                     .fixedBackoff(params.requestInterval.dividedBy(BACKOFF_INTERVAL_FACTOR))
  69                     .jitter(Jitter.random())
  70     )
  71
  72     override fun invoke(): Flux<CollectorConfiguration> =
  73             Flux.interval(firstRequestDelay, requestInterval)
  74                     .concatMap { askForConfig() }
  75                     .flatMap(::filterDifferentValues)
  76                     .map(::parseJsonResponse)
  77                     .map(::createCollectorConfiguration)
  78                     .retryWhen(retry)
  79
  80     private fun askForConfig(): Mono<String> = http.get(url)
  81
  82     private fun filterDifferentValues(configurationString: String) =
  83             hashOf(configurationString).let {
  84                 if (it == lastConfigurationHash.get()) {
  85                     Mono.empty()
  86                 } else {
  87                     lastConfigurationHash.set(it)
  88                     Mono.just(configurationString)
  89                 }
  90             }
  91
  92     private fun hashOf(str: String) = str.hashCode()
  93
  94     private fun parseJsonResponse(responseString: String): JsonObject =
  95             Json.createReader(StringReader(responseString)).readObject()
  96
  97     private fun createCollectorConfiguration(configuration: JsonObject): CollectorConfiguration {
  98         logger.info { "Obtained new configuration from consul:\n${configuration}" }
  99         val routing = configuration.getJsonArray("collector.routing")
 100
 101         return CollectorConfiguration(
 102                 kafkaBootstrapServers = configuration.getString("dmaap.kafkaBootstrapServers"),
 103                 routing = org.onap.dcae.collectors.veshv.model.routing {
 104                     for (route in routing) {
 105                         val routeObj = route.asJsonObject()
 106                         defineRoute {
 107                             fromDomain(routeObj.getString("fromDomain"))
 108                             toTopic(routeObj.getString("toTopic"))
 109                             withFixedPartitioning()
 110                         }
 111                     }
 112                 }.build()
 113         )
 114     }
 115
 116     companion object {
 117         private const val MAX_RETRIES = 5L
 118         private const val BACKOFF_INTERVAL_FACTOR = 30L
 119         private val logger = Logger(ConsulConfigurationProvider::class)
 120     }
 121 }
 122