Merge configurations

- changed temporarily HV-VES default log level to DEBUG as in current
implementation we are applying LogLevel defined in configuration file only
if we successfully retrieve one from configuration-module, which means
that inside of this module we are logging on default level (from logback
file). This should be fixed in future work
- reduced log level on SDK's CbsClientImpl as it's logging frequency was
too high

Change-Id: If50df18df099c34bfc36d39b045140f9b9ad87f6
Issue-ID: DCAEGEN2-1347
Signed-off-by: Filip Krzywka <filip.krzywka@nokia.com>

8 files changed, 92 insertions, 79 deletions

diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt
index 9684484b..ccce62a4 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt
@@ -19,6 +19,7 @@
  */
 package org.onap.dcae.collectors.veshv.config.api
 
+import arrow.core.getOrElse
 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
@@ -27,41 +28,56 @@ 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.config.impl.PartialConfiguration
 import org.onap.dcae.collectors.veshv.utils.arrow.throwOnLeft
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
 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
+import reactor.core.publisher.Mono
 
 class ConfigurationModule {
 
     private val cmd = HvVesCommandLineParser()
     private val configReader = FileConfigurationReader()
     private val configValidator = ConfigurationValidator()
+    private val merger = ConfigurationMerger()
 
     fun healthCheckPort(args: Array<String>): Int = cmd.getHealthcheckPort(args)
 
     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) }
+            Mono.just(cmd.getConfigurationFile(args))
+                    .throwOnLeft(::MissingArgumentException)
+                    .map {
+                        logger.info { "Using base configuration file: ${it.absolutePath}" }
+                        it.reader().use(configReader::loadConfig)
+                    }
                     .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()
+                    .flatMapMany { basePartialConfig ->
+                        cbsConfigurationProvider(basePartialConfig, configStateListener, mdc)
+                                .invoke()
                                 .map { merger.merge(basePartialConfig, it) }
-                                .map { configValidator.validate(it) }
-                                .throwOnLeft { ValidationException(it.message) }
+                                .map(configValidator::validate)
+                                .throwOnLeft()
                     }
 
+    private fun cbsConfigurationProvider(basePartialConfig: PartialConfiguration,
+                                         configStateListener: ConfigurationStateListener,
+                                         mdc: MappedDiagnosticContext): CbsConfigurationProvider =
+            CbsConfigurationProvider(
+                    CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
+                    cbsConfigurationFrom(basePartialConfig),
+                    configStateListener,
+                    mdc)
+
+    private fun cbsConfigurationFrom(basePartialConfig: PartialConfiguration) =
+            configValidator.validatedCbsConfiguration(basePartialConfig)
+                    .getOrElse { throw ValidationException("Invalid CBS section defined in configuration file") }
+
+    companion object {
+        private val logger = Logger(ConfigurationModule::class)
+    }
 }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/exceptions.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/exceptions.kt
index 2fc29829..bea7cc56 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/exceptions.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/exceptions.kt
@@ -19,6 +19,8 @@
  */
 package org.onap.dcae.collectors.veshv.config.api.model
 
-class MissingArgumentException(message: String, cause: Throwable?) : RuntimeException(message, cause)
+import org.onap.dcae.collectors.veshv.commandline.WrongArgumentError
+
+class MissingArgumentException(err: WrongArgumentError) : RuntimeException(err.message, err.cause)
 
 class ValidationException(message: String) : RuntimeException(message)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt
index f044492c..63d590a2 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt
@@ -74,8 +74,6 @@ internal class ConfigurationMerger {
                                      updateOption: Option<PartialCollectorConfig>) =
             applyUpdate(baseOption, updateOption) { base, update ->
                 PartialCollectorConfig(
-                        base.maxRequestSizeBytes.updateToGivenOrNone(update.maxRequestSizeBytes),
-                        base.kafkaServers.updateToGivenOrNone(update.kafkaServers),
                         base.routing.updateToGivenOrNone(update.routing)
                 )
             }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt
index 3e599b58..ead5655a 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt
@@ -27,13 +27,13 @@ import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
 import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
 import org.onap.dcae.collectors.veshv.config.api.model.HvVesConfiguration
 import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.config.api.model.ValidationException
 import org.onap.dcae.collectors.veshv.ssl.boundary.SecurityConfiguration
 import org.onap.dcae.collectors.veshv.utils.arrow.OptionUtils.binding
 import org.onap.dcae.collectors.veshv.utils.arrow.mapBinding
+import org.onap.dcae.collectors.veshv.utils.arrow.doOnEmpty
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import java.net.InetSocketAddress
-import java.time.Duration
 
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
@@ -41,33 +41,35 @@ import java.time.Duration
  */
 internal class ConfigurationValidator {
 
-    fun validate(partialConfig: PartialConfiguration)
-            : Either<ValidationError, HvVesConfiguration> = binding {
-        val logLevel = determineLogLevel(partialConfig.logLevel)
+    fun validate(partialConfig: PartialConfiguration) =
+            logger.info { "About to validate configuration: $partialConfig" }.let {
+                binding {
+                    val logLevel = determineLogLevel(partialConfig.logLevel)
 
-        val serverConfiguration = partialConfig.server.bind()
-                .let { createServerConfiguration(it).bind() }
+                    val serverConfiguration = validatedServerConfiguration(partialConfig)
+                            .doOnEmpty { logger.debug { "Cannot bind server configuration" } }
+                            .bind()
 
-        val cbsConfiguration = partialConfig.cbs.bind()
-                .let { createCbsConfiguration(it).bind() }
+                    val cbsConfiguration = validatedCbsConfiguration(partialConfig)
+                            .doOnEmpty { logger.debug { "Cannot bind cbs configuration" } }
+                            .bind()
 
-        val securityConfiguration = SecurityConfiguration(partialConfig.security.bind().keys)
+                    val securityConfiguration = SecurityConfiguration(partialConfig.security.bind().keys)
 
-// TOD0: retrieve when ConfigurationMerger is implemented
-//        val collectorConfiguration = partialConfig.collector.bind()
-//                .let { createCollectorConfig(it).bind() }
+                    val collectorConfiguration = validatedCollectorConfig(partialConfig)
+                            .doOnEmpty { logger.debug { "Cannot bind collector configuration" } }
+                            .bind()
+
+                    HvVesConfiguration(
+                            serverConfiguration,
+                            cbsConfiguration,
+                            securityConfiguration,
+                            collectorConfiguration,
+                            logLevel
+                    )
+                }.toEither { ValidationException("Some required configuration options are missing") }
+            }
 
-        HvVesConfiguration(
-                serverConfiguration,
-                cbsConfiguration,
-                securityConfiguration,
-// TOD0: swap when ConfigurationMerger is implemented
-//                    collectorConfiguration
-                CollectorConfiguration(emptyList()),
-// end TOD0
-                logLevel
-        )
-    }.toEither { ValidationError("Some required configuration options are missing") }
 
     private fun determineLogLevel(logLevel: Option<LogLevel>) =
             logLevel.getOrElse {
@@ -78,40 +80,38 @@ internal class ConfigurationValidator {
                 DEFAULT_LOG_LEVEL
             }
 
-    private fun createServerConfiguration(partial: PartialServerConfig) =
+    private fun validatedServerConfiguration(partial: PartialConfiguration) =
             partial.mapBinding {
-                ServerConfiguration(
-                        it.listenPort.bind(),
-                        it.idleTimeoutSec.bind(),
-                        it.maxPayloadSizeBytes.bind()
-                )
+                partial.server.bind().let {
+                    ServerConfiguration(
+                            it.listenPort.bind(),
+                            it.idleTimeoutSec.bind(),
+                            it.maxPayloadSizeBytes.bind()
+                    )
+                }
             }
 
-    private fun createCbsConfiguration(partial: PartialCbsConfig) =
+    fun validatedCbsConfiguration(partial: PartialConfiguration) =
             partial.mapBinding {
-                CbsConfiguration(
-                        it.firstRequestDelaySec.bind(),
-                        it.requestIntervalSec.bind()
-                )
+                it.cbs.bind().let {
+                    CbsConfiguration(
+                            it.firstRequestDelaySec.bind(),
+                            it.requestIntervalSec.bind()
+                    )
+                }
             }
 
-// TOD0: retrieve when ConfigurationMerger is implemented
-//    private fun createCollectorConfig(partial: PartialCollectorConfig) =
-//            partial.mapBinding {
-//                CollectorConfiguration(
-//                        it.maxRequestSizeBytes.bind(),
-//                        toKafkaServersString(it.kafkaServers.bind()),
-//                        it.routing.bind()
-//                )
-//            }
-
-    private fun toKafkaServersString(kafkaServers: List<InetSocketAddress>): String =
-            kafkaServers.joinToString(",") { "${it.hostName}:${it.port}" }
+    private fun validatedCollectorConfig(partial: PartialConfiguration) =
+            partial.mapBinding {
+                partial.collector.bind().let {
+                    CollectorConfiguration(
+                            it.routing.bind()
+                    )
+                }
+            }
 
     companion object {
         val DEFAULT_LOG_LEVEL = LogLevel.INFO
         private val logger = Logger(ConfigurationValidator::class)
     }
 }
-
-data class ValidationError(val message: String, val cause: Option<Throwable> = None)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt
index 9513107b..f6ae5bec 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt
@@ -24,6 +24,7 @@ import com.google.gson.GsonBuilder
 import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.DurationOfSecondsAdapter
 import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.OptionAdapter
 import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.SecurityAdapter
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
 
 import java.io.Reader
 import java.time.Duration
@@ -41,4 +42,9 @@ internal class FileConfigurationReader {
 
     fun loadConfig(input: Reader): PartialConfiguration =
             gson.fromJson(input, PartialConfiguration::class.java)
+                    .also { logger.info { "Successfully read file and parsed json to configuration: $it" } }
+
+    companion object {
+        private val logger = Logger(FileConfigurationReader::class)
+    }
 }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
index c1a98294..c6730a4c 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
@@ -45,7 +45,7 @@ internal class HvVesCommandLineParser(private val parser: CommandLineParser = De
                 it.stringValue(CONFIGURATION_FILE).map(::File)
             }.toEither {
                 WrongArgumentError(
-                        message = "Unexpected error when parsing command line arguments",
+                        message = "Base configuration filepath missing on command line",
                         cmdLineOptionsList = cmdLineOptionsList)
             }
 
@@ -53,8 +53,7 @@ internal class HvVesCommandLineParser(private val parser: CommandLineParser = De
             parse(args) {
                 it.intValue(HEALTH_CHECK_API_PORT)
             }.getOrElse {
-                logger.info { "Healthcheck port missing on command line," +
-                        " using default: $DEFAULT_HEALTHCHECK_PORT" }
+                logger.info { "Healthcheck port missing on command line, using default: $DEFAULT_HEALTHCHECK_PORT" }
                 DEFAULT_HEALTHCHECK_PORT
             }
 
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/partial_configuration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/partial_configuration.kt
index f3c149cd..b4e1bf6b 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/partial_configuration.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/partial_configuration.kt
@@ -53,7 +53,5 @@ internal data class PartialCbsConfig(
 internal data class PartialSecurityConfig(val keys: Option<SecurityKeys> = None)
 
 internal data class PartialCollectorConfig(
-        val maxRequestSizeBytes: Option<Int> = None,
-        val kafkaServers: Option<List<InetSocketAddress>> = None, // TOD0: remove properties and simplify this part
         val routing: Option<Routing> = None
 )
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
index 4b89488b..55d06cdd 100644
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
@@ -64,8 +64,6 @@ internal object ConfigurationValidatorTest : Spek({
                             Some(mock())
                     )),
                     Some(PartialCollectorConfig(
-                            Some(4),
-                            Some(emptyList()),
                             someFromEmptyRouting
                     )),
                     None
@@ -103,8 +101,6 @@ internal object ConfigurationValidatorTest : Spek({
                             securityKeys
                     )),
                     Some(PartialCollectorConfig(
-                            Some(4),
-                            Some(emptyList()),
                             someFromEmptyRouting
                     )),
                     Some(LogLevel.INFO)
@@ -152,8 +148,6 @@ internal object ConfigurationValidatorTest : Spek({
                             securityKeys
                     )),
                     Some(PartialCollectorConfig(
-                            Some(4),
-                            Some(emptyList()),
                             someFromEmptyRouting
                     )),
                     Some(LogLevel.INFO)