Remove environment variables and program arguments

- Move all command line program arguments to json file.
- Reorganize configuration classes and the way they are passed through application
- Implement HV VES configuration stream
- Create concrete configuration from partial one
- Modify main HV-VES server starting pipeline

Change-Id: I6cf874b6904ed768e4820b8132f5f760299c929e
Signed-off-by: Jakub Dudycz <jakub.dudycz@nokia.com>
Issue-ID: DCAEGEN2-1340

9 files changed, 204 insertions, 243 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 874cc5b0..5b547a28 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,13 +19,25 @@
  */
 package org.onap.dcae.collectors.veshv.config.api
 
-import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
-import org.onap.dcae.collectors.veshv.config.impl.ArgVesHvConfiguration
+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.ArgHvVesConfiguration
+import org.onap.dcae.collectors.veshv.config.impl.ConfigurationValidator
+import org.onap.dcae.collectors.veshv.config.impl.FileConfigurationReader
+import org.onap.dcae.collectors.veshv.utils.arrow.throwOnLeft
 import reactor.core.publisher.Flux
 
 class ConfigurationModule {
-    fun createConfigurationFromCommandLine(args: Array<String>) =
-            ArgVesHvConfiguration().parse(args)
 
-    fun hvVesConfigurationUpdates(): Flux<ServerConfiguration> = Flux.never<ServerConfiguration>()
+    private val cmd = ArgHvVesConfiguration()
+    private val configReader = FileConfigurationReader()
+    private val configValidator = ConfigurationValidator()
+
+    fun hvVesConfigurationUpdates(args: Array<String>): Flux<HvVesConfiguration> =
+            Flux.just(cmd.parse(args))
+                    .throwOnLeft { MissingArgumentException(it.message, it.cause) }
+                    .map { configReader.loadConfig(it.reader()) }
+                    .map { configValidator.validate(it) }
+                    .throwOnLeft { ValidationException(it.message) }
 }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ServerConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Configuration.kt
index 8fbc649b..566f2c08 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ServerConfiguration.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Configuration.kt
@@ -21,22 +21,34 @@ package org.onap.dcae.collectors.veshv.config.api.model
 
 import org.onap.dcae.collectors.veshv.ssl.boundary.SecurityConfiguration
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
-import java.net.InetSocketAddress
 import java.time.Duration
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since May 2018
  */
+data class HvVesConfiguration(
+        val server: ServerConfiguration,
+        val cbs: CbsConfiguration,
+        val security: SecurityConfiguration,
+        val collector: CollectorConfiguration,
+        val logLevel: LogLevel
+)
+
 data class ServerConfiguration(
-        val serverListenAddress: InetSocketAddress,
-        val kafkaConfiguration: KafkaConfiguration,
-        val configurationProviderParams: ConfigurationProviderParams,
-        val securityConfiguration: SecurityConfiguration,
+        val listenPort: Int,
         val idleTimeout: Duration,
-        val healthCheckApiListenAddress: InetSocketAddress,
-        val maximumPayloadSizeBytes: Int,
-        val logLevel: LogLevel,
-        val dummyMode: Boolean = false
+        val maxPayloadSizeBytes: Int
 )
 
+data class CbsConfiguration(
+        val firstRequestDelay: Duration,
+        val requestInterval: Duration
+)
+
+data class CollectorConfiguration(
+        val maxRequestSizeBytes: Int,
+        val kafkaServers: String,
+        val routing: Routing,
+        val dummyMode: Boolean = false
+)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/CollectorConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Exceptions.kt
index 7999451e..2fc29829 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/CollectorConfiguration.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Exceptions.kt
@@ -2,7 +2,7 @@
  * ============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.
@@ -19,8 +19,6 @@
  */
 package org.onap.dcae.collectors.veshv.config.api.model
 
-/**
- * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
- * @since May 2018
- */
-data class CollectorConfiguration(val routing: Routing)
+class MissingArgumentException(message: String, cause: Throwable?) : RuntimeException(message, cause)
+
+class ValidationException(message: String) : RuntimeException(message)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt
deleted file mode 100644
index f9dff203..00000000
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * ============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.config.api.model
-
-/**
- * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
- * @since December 2018
- */
-data class KafkaConfiguration(val bootstrapServers: String, val maximalRequestSizeBytes: Int)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/routing.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt
index 847f35ad..aab8ecad 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/routing.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA
+ * 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.
@@ -30,7 +30,7 @@ data class Routing(val routes: List<Route>) {
             Option.fromNullable(routes.find { it.applies(commonHeader) })
 }
 
-data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int = {0}) {
+data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int = { 0 }) {
 
     fun applies(commonHeader: CommonEventHeader) = commonHeader.domain == domain
 
@@ -40,24 +40,17 @@ data class Route(val domain: String, val targetTopic: String, val partitioning:
 
 
 /*
-Configuration DSL
- */
+HvVesConfiguration DSL
+*/
 
-fun routing(init: RoutingBuilder.() -> Unit): RoutingBuilder {
-    val conf = RoutingBuilder()
-    conf.init()
-    return conf
-}
+fun routing(init: RoutingBuilder.() -> Unit): RoutingBuilder = RoutingBuilder().apply(init)
 
 class RoutingBuilder {
     private val routes: MutableList<RouteBuilder> = mutableListOf()
 
-    fun defineRoute(init: RouteBuilder.() -> Unit): RouteBuilder {
-        val rule = RouteBuilder()
-        rule.init()
-        routes.add(rule)
-        return rule
-    }
+    fun defineRoute(init: RouteBuilder.() -> Unit): RouteBuilder = RouteBuilder()
+            .apply(init)
+            .also { routes.add(it) }
 
     fun build() = Routing(routes.map { it.build() }.toList())
 }
@@ -68,18 +61,17 @@ class RouteBuilder {
     private lateinit var targetTopic: String
     private lateinit var partitioning: (CommonEventHeader) -> Int
 
-    fun fromDomain(domain: String) : RouteBuilder = apply {
+    fun fromDomain(domain: String): RouteBuilder = apply {
         this.domain = domain
     }
 
-    fun toTopic(targetTopic: String) : RouteBuilder = apply {
+    fun toTopic(targetTopic: String): RouteBuilder = apply {
         this.targetTopic = targetTopic
     }
 
-    fun withFixedPartitioning(num: Int = 0) : RouteBuilder = apply {
+    fun withFixedPartitioning(num: Int = 0): RouteBuilder = apply {
         partitioning = { num }
     }
 
     fun build() = Route(domain, targetTopic, partitioning)
-
 }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ConfigurationProviderParams.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgHvVesConfiguration.kt
index 847279d9..9587d5b0 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ConfigurationProviderParams.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgHvVesConfiguration.kt
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018-2019 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.
@@ -17,13 +17,20 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.collectors.veshv.config.api.model
+package org.onap.dcae.collectors.veshv.config.impl
 
-import java.time.Duration
+import arrow.core.Option
+import org.apache.commons.cli.CommandLine
+import org.apache.commons.cli.DefaultParser
+import org.onap.dcae.collectors.veshv.commandline.ArgBasedConfiguration
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_FILE
+import org.onap.dcae.collectors.veshv.commandline.stringValue
+import java.io.File
 
-/**
- * @author Jakub Dudycz <jakub.dudycz@nokia.com>
- * @since July 2018
- */
-data class ConfigurationProviderParams(val firstRequestDelay: Duration,
-                                       val requestInterval: Duration)
+internal class ArgHvVesConfiguration : ArgBasedConfiguration<File>(DefaultParser()) {
+    override val cmdLineOptionsList = listOf(CONFIGURATION_FILE)
+
+    override fun getConfiguration(cmdLine: CommandLine): Option<File> =
+            cmdLine.stringValue(CONFIGURATION_FILE).map(::File)
+
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt
deleted file mode 100644
index 08346d30..00000000
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * ============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.config.impl
-
-import arrow.core.Option
-import arrow.core.fix
-import arrow.core.getOrElse
-import arrow.instances.option.monad.monad
-import arrow.typeclasses.binding
-import org.apache.commons.cli.CommandLine
-import org.apache.commons.cli.DefaultParser
-import org.onap.dcae.collectors.veshv.commandline.ArgBasedConfiguration
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_REQUEST_INTERVAL
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_FIRST_REQUEST_DELAY
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.DUMMY_MODE
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.HEALTH_CHECK_API_PORT
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.IDLE_TIMEOUT_SEC
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KAFKA_SERVERS
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KEY_STORE_FILE
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KEY_STORE_PASSWORD
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.LISTEN_PORT
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.LOG_LEVEL
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.MAXIMUM_PAYLOAD_SIZE_BYTES
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.SSL_DISABLE
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.TRUST_STORE_FILE
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.TRUST_STORE_PASSWORD
-import org.onap.dcae.collectors.veshv.commandline.hasOption
-import org.onap.dcae.collectors.veshv.commandline.intValue
-import org.onap.dcae.collectors.veshv.commandline.longValue
-import org.onap.dcae.collectors.veshv.commandline.stringValue
-import org.onap.dcae.collectors.veshv.config.api.model.ConfigurationProviderParams
-import org.onap.dcae.collectors.veshv.config.api.model.KafkaConfiguration
-import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
-import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
-import org.onap.dcae.collectors.veshv.ssl.boundary.createSecurityConfiguration
-import org.onap.dcae.collectors.veshv.utils.arrow.doOnFailure
-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
-
-
-internal class ArgVesHvConfiguration : ArgBasedConfiguration<ServerConfiguration>(DefaultParser()) {
-    override val cmdLineOptionsList = listOf(
-            KAFKA_SERVERS,
-            HEALTH_CHECK_API_PORT,
-            LISTEN_PORT,
-            CONFIGURATION_FIRST_REQUEST_DELAY,
-            CONFIGURATION_REQUEST_INTERVAL,
-            SSL_DISABLE,
-            KEY_STORE_FILE,
-            KEY_STORE_PASSWORD,
-            TRUST_STORE_FILE,
-            TRUST_STORE_PASSWORD,
-            IDLE_TIMEOUT_SEC,
-            MAXIMUM_PAYLOAD_SIZE_BYTES,
-            DUMMY_MODE,
-            LOG_LEVEL
-    )
-
-    override fun getConfiguration(cmdLine: CommandLine): Option<ServerConfiguration> =
-            Option.monad().binding {
-                val healthCheckApiPort = cmdLine.intValue(
-                        HEALTH_CHECK_API_PORT,
-                        DefaultValues.HEALTH_CHECK_API_PORT
-                )
-                val kafkaServers = cmdLine.stringValue(KAFKA_SERVERS).bind()
-                val listenPort = cmdLine.intValue(LISTEN_PORT).bind()
-                val idleTimeoutSec = cmdLine.longValue(IDLE_TIMEOUT_SEC, DefaultValues.IDLE_TIMEOUT_SEC)
-                val maxPayloadSizeBytes = cmdLine.intValue(
-                        MAXIMUM_PAYLOAD_SIZE_BYTES,
-                        DefaultValues.MAX_PAYLOAD_SIZE_BYTES
-                )
-                val dummyMode = cmdLine.hasOption(DUMMY_MODE)
-                val security = createSecurityConfiguration(cmdLine)
-                        .doOnFailure { ex ->
-                            logger.withError { log("Could not read security keys", ex) }
-                        }
-                        .toOption()
-                        .bind()
-                val logLevel = cmdLine.stringValue(LOG_LEVEL, DefaultValues.LOG_LEVEL)
-                val configurationProviderParams = createConfigurationProviderParams(cmdLine).bind()
-                ServerConfiguration(
-                        serverListenAddress = InetSocketAddress(listenPort),
-                        kafkaConfiguration = KafkaConfiguration(kafkaServers, maxPayloadSizeBytes),
-                        healthCheckApiListenAddress = InetSocketAddress(healthCheckApiPort),
-                        configurationProviderParams = configurationProviderParams,
-                        securityConfiguration = security,
-                        idleTimeout = Duration.ofSeconds(idleTimeoutSec),
-                        maximumPayloadSizeBytes = maxPayloadSizeBytes,
-                        dummyMode = dummyMode,
-                        logLevel = determineLogLevel(logLevel)
-                )
-            }.fix()
-
-    private fun createConfigurationProviderParams(cmdLine: CommandLine): Option<ConfigurationProviderParams> =
-            Option.monad().binding {
-                val firstRequestDelay = cmdLine.longValue(
-                        CONFIGURATION_FIRST_REQUEST_DELAY,
-                        DefaultValues.CONFIGURATION_FIRST_REQUEST_DELAY
-                )
-                val requestInterval = cmdLine.longValue(
-                        CONFIGURATION_REQUEST_INTERVAL,
-                        DefaultValues.CONFIGURATION_REQUEST_INTERVAL
-                )
-                ConfigurationProviderParams(
-                        Duration.ofSeconds(firstRequestDelay),
-                        Duration.ofSeconds(requestInterval)
-                )
-            }.fix()
-
-    private fun determineLogLevel(logLevel: String) = LogLevel.optionFromString(logLevel)
-            .getOrElse {
-                logger.warn {
-                    "Failed to parse $logLevel as $LOG_LEVEL command line. " +
-                            "Using default log level (${DefaultValues.LOG_LEVEL})"
-                }
-                LogLevel.valueOf(DefaultValues.LOG_LEVEL)
-            }
-
-
-    internal object DefaultValues {
-        const val HEALTH_CHECK_API_PORT = 6060
-        const val CONFIGURATION_FIRST_REQUEST_DELAY = 10L
-        const val CONFIGURATION_REQUEST_INTERVAL = 5L
-        const val IDLE_TIMEOUT_SEC = 60L
-        const val MAX_PAYLOAD_SIZE_BYTES = WireFrameMessage.DEFAULT_MAX_PAYLOAD_SIZE_BYTES
-        val LOG_LEVEL = LogLevel.INFO.name
-    }
-
-    companion object {
-        private val logger = Logger(ArgVesHvConfiguration::class)
-    }
-}
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
new file mode 100644
index 00000000..6c74c33f
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt
@@ -0,0 +1,118 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * 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.
+ * 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.config.impl
+
+import arrow.core.Either
+import arrow.core.None
+import arrow.core.Option
+import arrow.core.Some
+import arrow.core.getOrElse
+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.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.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>
+ * @since March 2019
+ */
+internal class ConfigurationValidator {
+
+    fun validate(partialConfig: PartialConfiguration)
+            : Either<ValidationError, HvVesConfiguration> = binding {
+        val logLevel = determineLogLevel(partialConfig.logLevel)
+
+        val serverConfiguration = partialConfig.server.bind()
+                .let { createServerConfiguration(it).bind() }
+
+        val cbsConfiguration = partialConfig.cbs.bind()
+                .let { createCbsConfiguration(it).bind() }
+
+        val securityConfiguration = partialConfig.security.bind()
+                .let { createSecurityConfiguration(it).bind() }
+
+        val collectorConfiguration = partialConfig.collector.bind()
+                .let { createCollectorConfig(it).bind() }
+
+        HvVesConfiguration(
+                serverConfiguration,
+                cbsConfiguration,
+                securityConfiguration,
+                collectorConfiguration,
+                logLevel
+        )
+    }.toEither { ValidationError("Some required configuration options are missing") }
+
+    private fun determineLogLevel(logLevel: Option<LogLevel>) =
+            logLevel.getOrElse {
+                logger.warn {
+                    "Missing or invalid \"logLevel\" field. " +
+                            "Using default log level ($DEFAULT_LOG_LEVEL)"
+                }
+                DEFAULT_LOG_LEVEL
+            }
+
+    private fun createServerConfiguration(partial: PartialServerConfig) =
+            partial.mapBinding {
+                ServerConfiguration(
+                        it.listenPort.bind(),
+                        Duration.ofSeconds(it.idleTimeoutSec.bind().toLong()),
+                        it.maxPayloadSizeBytes.bind()
+                )
+            }
+
+    private fun createCbsConfiguration(partial: PartialCbsConfig) =
+            partial.mapBinding {
+                CbsConfiguration(
+                        Duration.ofSeconds(it.firstRequestDelaySec.bind().toLong()),
+                        Duration.ofSeconds(it.requestIntervalSec.bind().toLong())
+                )
+            }
+
+    private fun createSecurityConfiguration(partial: PartialSecurityConfig) =
+            partial.keys.map { SecurityConfiguration(Some(it)) }
+
+    private fun createCollectorConfig(partial: PartialCollectorConfig) =
+            partial.mapBinding {
+                CollectorConfiguration(
+                        it.maxRequestSizeBytes.bind(),
+                        toKafkaServersString(it.kafkaServers.bind()),
+                        it.routing.bind(),
+                        it.dummyMode.bind()
+                )
+            }
+
+    private fun toKafkaServersString(kafkaServers: List<InetSocketAddress>): String =
+            kafkaServers.joinToString(",") { "${it.hostName}:${it.port}" }
+
+    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/PartialConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt
index 8d9cca73..3e6df3e0 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt
@@ -19,6 +19,7 @@
  */
 package org.onap.dcae.collectors.veshv.config.impl
 
+import arrow.core.None
 import arrow.core.Option
 import org.onap.dcae.collectors.veshv.config.api.model.Routing
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
@@ -30,31 +31,29 @@ import java.net.InetSocketAddress
  * @since February 2019
  */
 internal data class PartialConfiguration(
-        val server : Option<PartialServerConfig>,
-        val cbs : Option<PartialCbsConfig>,
-        val security : Option<PartialSecurityConfig>,
-        val kafka : Option<PartialKafkaConfig>,
-        val logLevel : Option<LogLevel>
+        val server: Option<PartialServerConfig> = None,
+        val cbs: Option<PartialCbsConfig> = None,
+        val security: Option<PartialSecurityConfig> = None,
+        val collector: Option<PartialCollectorConfig> = None,
+        val logLevel: Option<LogLevel> = None
 )
 
 internal data class PartialServerConfig(
-        val healthCheckApiPort : Option<Int>,
-        val listenPort : Option<Int>,
-        val idleTimeoutSec : Option<Int>,
-        val maximumPayloadSizeBytes : Option<Int>,
-        val dummyMode : Option<Boolean>
+        val listenPort: Option<Int> = None,
+        val idleTimeoutSec: Option<Int> = None,
+        val maxPayloadSizeBytes: Option<Int> = None
 )
 
 internal data class PartialCbsConfig(
-        val firstRequestDelaySec : Option<Int>,
-        val requestIntervalSec : Option<Int>
+        val firstRequestDelaySec: Option<Int> = None,
+        val requestIntervalSec: Option<Int> = None
 )
 
-internal data class PartialSecurityConfig(
-        val sslDisable : Option<Boolean>,
-        val keys : Option<SecurityKeys>)
+internal data class PartialSecurityConfig(val keys: Option<SecurityKeys> = None)
 
-internal data class PartialKafkaConfig(
-    val kafkaServers : Option<Array<InetSocketAddress>>,
-    val routing : Option<Routing>
+internal data class PartialCollectorConfig(
+        val dummyMode: Option<Boolean> = None,
+        val maxRequestSizeBytes: Option<Int> = None,
+        val kafkaServers: Option<List<InetSocketAddress>> = None,
+        val routing: Option<Routing> = None
 )