diff options
Diffstat (limited to 'sources/hv-collector-configuration/src/main/kotlin')
7 files changed, 92 insertions, 73 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 ) |