Merge "Extract transforming logic from validator"

7 files changed, 273 insertions, 141 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 f0ee3a42..ded75838 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,12 +19,11 @@
  */
 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
 import org.onap.dcae.collectors.veshv.config.impl.CbsConfigurationProvider
 import org.onap.dcae.collectors.veshv.config.impl.ConfigurationMerger
+import org.onap.dcae.collectors.veshv.config.impl.ConfigurationTransformer
 import org.onap.dcae.collectors.veshv.config.impl.ConfigurationValidator
 import org.onap.dcae.collectors.veshv.config.impl.HvVesCommandLineParser
 import org.onap.dcae.collectors.veshv.config.impl.JsonConfigurationParser
@@ -41,8 +40,9 @@ class ConfigurationModule {
 
     private val cmd = HvVesCommandLineParser()
     private val configParser = JsonConfigurationParser()
+    private val configMerger = ConfigurationMerger()
     private val configValidator = ConfigurationValidator()
-    private val merger = ConfigurationMerger()
+    private val configTransformer = ConfigurationTransformer()
 
     fun healthCheckPort(args: Array<String>): Int = cmd.getHealthcheckPort(args)
 
@@ -58,14 +58,15 @@ class ConfigurationModule {
                     .flatMapMany { basePartialConfig ->
                         cbsConfigurationProvider(basePartialConfig, configStateListener, mdc)
                                 .invoke()
-                                .map { merger.merge(basePartialConfig, it) }
+                                .map { configMerger.merge(basePartialConfig, it) }
                                 .map(configValidator::validate)
                                 .throwOnLeft()
+                                .map(configTransformer::toFinalConfiguration)
                     }
 
     private fun cbsConfigurationProvider(basePartialConfig: PartialConfiguration,
                                          configStateListener: ConfigurationStateListener,
-                                         mdc: MappedDiagnosticContext): CbsConfigurationProvider =
+                                         mdc: MappedDiagnosticContext) =
             CbsConfigurationProvider(
                     CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()),
                     cbsConfigurationFrom(basePartialConfig),
@@ -73,11 +74,12 @@ class ConfigurationModule {
                     configStateListener,
                     mdc)
 
-    private fun cbsConfigurationFrom(basePartialConfig: PartialConfiguration) = configValidator
-            .validatedCbsConfiguration(basePartialConfig)
-            .getOrElse { throw ValidationException("Invalid CBS section defined in configuration file") }
+    private fun cbsConfigurationFrom(basePartialConfig: PartialConfiguration) =
+            configValidator.validatedCbsConfiguration(basePartialConfig)
+                    .let { configTransformer.toCbsConfiguration(it) }
 
     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/configuration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/configuration.kt
index 8db2f770..fd3cccd9 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/configuration.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/configuration.kt
@@ -46,13 +46,6 @@ data class CbsConfiguration(
 )
 
 data class CollectorConfiguration(
-        val routing: Routing
-) {
-    val maxPayloadSizeBytes by lazy {
-        routing.map { it.sink.maxPayloadSizeBytes() }.max() ?: DEFAULT_MAX_PAYLOAD_SIZE
-    }
-
-    companion object {
-        internal const val DEFAULT_MAX_PAYLOAD_SIZE = 1024 * 1024
-    }
-}
+        val routing: Routing,
+        val maxPayloadSizeBytes: Int
+)
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 e6707825..96fa4213 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
@@ -29,24 +29,23 @@ import arrow.core.toOption
  * @since March 2019
  */
 internal class ConfigurationMerger {
-    fun merge(base: PartialConfiguration, update: PartialConfiguration): PartialConfiguration =
-            PartialConfiguration(
-                    listenPort = base.listenPort.updateToGivenOrNone(update.listenPort),
-                    idleTimeoutSec = base.idleTimeoutSec.updateToGivenOrNone(update.idleTimeoutSec),
+    fun merge(base: PartialConfiguration, update: PartialConfiguration) = PartialConfiguration(
+            listenPort = base.listenPort.updateToGivenOrNone(update.listenPort),
+            idleTimeoutSec = base.idleTimeoutSec.updateToGivenOrNone(update.idleTimeoutSec),
 
-                    firstRequestDelaySec = base.firstRequestDelaySec.updateToGivenOrNone(update.firstRequestDelaySec),
-                    requestIntervalSec = base.requestIntervalSec.updateToGivenOrNone(update.requestIntervalSec),
+            firstRequestDelaySec = base.firstRequestDelaySec.updateToGivenOrNone(update.firstRequestDelaySec),
+            requestIntervalSec = base.requestIntervalSec.updateToGivenOrNone(update.requestIntervalSec),
 
-                    sslDisable = base.sslDisable.updateToGivenOrNone(update.sslDisable),
-                    keyStoreFile = base.keyStoreFile.updateToGivenOrNone(update.keyStoreFile),
-                    keyStorePasswordFile = base.keyStorePasswordFile.updateToGivenOrNone(update.keyStorePasswordFile),
-                    trustStoreFile = base.trustStoreFile.updateToGivenOrNone(update.trustStoreFile),
-                    trustStorePasswordFile = base.trustStorePasswordFile.updateToGivenOrNone(update.trustStorePasswordFile),
+            sslDisable = base.sslDisable.updateToGivenOrNone(update.sslDisable),
+            keyStoreFile = base.keyStoreFile.updateToGivenOrNone(update.keyStoreFile),
+            keyStorePasswordFile = base.keyStorePasswordFile.updateToGivenOrNone(update.keyStorePasswordFile),
+            trustStoreFile = base.trustStoreFile.updateToGivenOrNone(update.trustStoreFile),
+            trustStorePasswordFile = base.trustStorePasswordFile.updateToGivenOrNone(update.trustStorePasswordFile),
 
-                    streamPublishers = base.streamPublishers.updateToGivenOrNone(update.streamPublishers),
+            streamPublishers = base.streamPublishers.updateToGivenOrNone(update.streamPublishers),
 
-                    logLevel = base.logLevel.updateToGivenOrNone(update.logLevel)
-            )
+            logLevel = base.logLevel.updateToGivenOrNone(update.logLevel)
+    )
 
     private fun <T> Option<T>.updateToGivenOrNone(update: Option<T>) =
             update.getOrElse(this::orNull).toOption()
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformer.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformer.kt
new file mode 100644
index 00000000..08cce136
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformer.kt
@@ -0,0 +1,116 @@
+/*
+ * ============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.None
+import arrow.core.Option
+import arrow.core.getOrElse
+import arrow.core.toOption
+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.Route
+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.logging.LogLevel
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeys
+import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeysStore
+import org.onap.dcaegen2.services.sdk.security.ssl.Passwords
+import java.nio.file.Paths
+import java.time.Duration
+
+internal class ConfigurationTransformer {
+
+    fun toFinalConfiguration(validatedConfig: ValidatedPartialConfiguration): HvVesConfiguration {
+        val serverConfiguration = toServerConfiguration(validatedConfig)
+
+        val cbsConfiguration = toCbsConfiguration(validatedConfig.cbsConfiguration)
+
+        val securityConfiguration = determineSecurityConfiguration(validatedConfig)
+
+        val collectorConfiguration = toCollectorConfiguration(validatedConfig)
+
+        val logLevel = determineLogLevel(validatedConfig.logLevel)
+
+        return HvVesConfiguration(
+                serverConfiguration,
+                cbsConfiguration,
+                securityConfiguration,
+                collectorConfiguration,
+                logLevel
+        )
+    }
+
+    fun toCbsConfiguration(cbsConfiguration: ValidatedCbsConfiguration) = CbsConfiguration(
+            Duration.ofSeconds(cbsConfiguration.firstRequestDelaySec),
+            Duration.ofSeconds(cbsConfiguration.requestIntervalSec)
+    )
+
+    private fun toServerConfiguration(validatedConfig: ValidatedPartialConfiguration) = ServerConfiguration(
+            validatedConfig.listenPort,
+            Duration.ofSeconds(validatedConfig.idleTimeoutSec)
+    )
+
+    private fun determineSecurityConfiguration(validConfig: ValidatedPartialConfiguration) =
+            validConfig.securityConfiguration.fold({ SecurityConfiguration(None) }, { createSecurityConfiguration(it) })
+
+    private fun toCollectorConfiguration(validatedConfig: ValidatedPartialConfiguration) =
+            validatedConfig.streamPublishers.map { Route(it.name(), it) }
+                    .let { routing ->
+                        CollectorConfiguration(
+                                routing,
+                                determineMaxPayloadSize(routing)
+                        )
+                    }
+
+    private fun createSecurityConfiguration(paths: ValidatedSecurityPaths) = SecurityConfiguration(
+            ImmutableSecurityKeys.builder()
+                    .keyStore(ImmutableSecurityKeysStore.of(Paths.get(paths.keyStoreFile)))
+                    .keyStorePassword(Passwords.fromPath(Paths.get(paths.keyStorePasswordFile)))
+                    .trustStore(ImmutableSecurityKeysStore.of(Paths.get(paths.trustStoreFile)))
+                    .trustStorePassword(Passwords.fromPath(Paths.get(paths.trustStorePasswordFile)))
+                    .build()
+                    .toOption()
+    )
+
+    private fun determineMaxPayloadSize(routing: List<Route>) =
+            routing.map { it.sink.maxPayloadSizeBytes() }.max() ?: useDefaultMaxPayloadSize()
+
+    private fun determineLogLevel(logLevel: Option<LogLevel>) =
+            logLevel.getOrElse(::useDefaultLogLevel)
+
+    private fun useDefaultMaxPayloadSize() = DEFAULT_MAX_PAYLOAD_SIZE.also {
+        logger.warn {
+            "Failed to determine \"maxPayloadSizeBytes\" field from routing. Using default ($it)"
+        }
+    }
+
+    private fun useDefaultLogLevel() = DEFAULT_LOG_LEVEL.also {
+        logger.warn { "Missing or invalid \"logLevel\" field. Using default log level ($it)" }
+    }
+
+    companion object {
+        private val logger = Logger(ConfigurationTransformer::class)
+
+        private val DEFAULT_LOG_LEVEL = LogLevel.INFO
+        private const val DEFAULT_MAX_PAYLOAD_SIZE = 1024 * 1024
+    }
+}
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 f4ce592f..c97c975f 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
@@ -19,29 +19,15 @@
  */
 package org.onap.dcae.collectors.veshv.config.impl
 
-import arrow.core.None
-import arrow.core.Option
-import arrow.core.Some
-import arrow.core.getOrElse
-import arrow.core.toOption
-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.Route
-import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
+import arrow.core.Either
+import arrow.core.Left
+import arrow.core.Right
+import arrow.data.Invalid
+import arrow.data.Validated
 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.doOnEmpty
-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.arrow.flatFold
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeys
-import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeysStore
-import org.onap.dcaegen2.services.sdk.security.ssl.Passwords
-import java.io.File
-import java.nio.file.Path
-import java.time.Duration
+
 
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
@@ -49,104 +35,62 @@ import java.time.Duration
  */
 internal class ConfigurationValidator {
 
-    fun validate(partialConfig: PartialConfiguration) =
-            logger.info { "About to validate configuration: $partialConfig" }.let {
-                binding {
-                    val logLevel = determineLogLevel(partialConfig.logLevel)
-
-                    val serverConfiguration = validatedServerConfiguration(partialConfig)
-                            .doOnEmpty { logger.debug { "Cannot bind server configuration" } }
-                            .bind()
-
-                    val cbsConfiguration = validatedCbsConfiguration(partialConfig)
-                            .doOnEmpty { logger.debug { "Cannot bind cbs configuration" } }
-                            .bind()
-
-                    val securityConfiguration = determineSecurityConfiguration(partialConfig)
-                            .doOnEmpty { logger.debug { "Cannot bind security configuration" } }
-                            .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") }
-            }
-
+    fun validate(partial: PartialConfiguration): Either<ValidationException, ValidatedPartialConfiguration> =
+            logger.info { "About to validate configuration: $partial" }.let {
+                val invalidFields = mutableSetOf(
+                        validate(partial::streamPublishers)
+                )
+                        .union(cbsConfigurationValidation(partial))
+                        .union(serverConfigurationValidation(partial))
+                        .union(securityValidation(partial))
+                        .filter { it.isInvalid }
 
-    private fun determineLogLevel(logLevel: Option<LogLevel>) =
-            logLevel.getOrElse {
-                logger.warn {
-                    "Missing or invalid \"logLevel\" field. " +
-                            "Using default log level ($DEFAULT_LOG_LEVEL)"
+                if (invalidFields.isNotEmpty()) {
+                    return Left(ValidationException(validationMessageFrom(invalidFields)))
                 }
-                DEFAULT_LOG_LEVEL
-            }
 
-    private fun validatedServerConfiguration(partial: PartialConfiguration) =
-            partial.mapBinding {
-                ServerConfiguration(
-                        it.listenPort.bind(),
-                        Duration.ofSeconds(it.idleTimeoutSec.bind())
-                )
+                Right(partial.unsafeAsValidated())
             }
 
-    internal fun validatedCbsConfiguration(partial: PartialConfiguration) =
-            partial.mapBinding {
-                CbsConfiguration(
-                        Duration.ofSeconds(it.firstRequestDelaySec.bind()),
-                        Duration.ofSeconds(it.requestIntervalSec.bind())
-                )
-            }
-
-    private fun determineSecurityConfiguration(partial: PartialConfiguration) =
-            partial.sslDisable.fold({ createSecurityConfiguration(partial) }, { sslDisabled ->
-                if (sslDisabled) {
-                    Some(SecurityConfiguration(None))
-                } else {
-                    createSecurityConfiguration(partial)
-                }
+    fun validatedCbsConfiguration(partial: PartialConfiguration) = ValidatedCbsConfiguration(
+            firstRequestDelaySec = getOrThrowValidationException(partial::firstRequestDelaySec),
+            requestIntervalSec = getOrThrowValidationException(partial::requestIntervalSec)
+    )
+
+    private fun cbsConfigurationValidation(partial: PartialConfiguration) = setOf(
+            validate(partial::firstRequestDelaySec),
+            validate(partial::requestIntervalSec)
+    )
+
+    private fun serverConfigurationValidation(partial: PartialConfiguration) = setOf(
+            validate(partial::listenPort),
+            validate(partial::idleTimeoutSec)
+    )
+
+    private fun securityValidation(partial: PartialConfiguration) =
+            partial.sslDisable.flatFold({
+                validatedSecurityConfiguration(partial)
+            }, {
+                setOf(Validated.Valid("sslDisable flag is set to true"))
             })
 
-    private fun createSecurityConfiguration(partial: PartialConfiguration): Option<SecurityConfiguration> =
-            partial.mapBinding {
-                SecurityConfiguration(
-                        createSecurityKeys(
-                                File(it.keyStoreFile.bind()).toPath(),
-                                File(it.keyStorePasswordFile.bind()).toPath(),
-                                File(it.trustStoreFile.bind()).toPath(),
-                                File(it.trustStorePasswordFile.bind()).toPath()
-                        ).toOption()
-                )
-            }
+    private fun validatedSecurityConfiguration(partial: PartialConfiguration) = setOf(
+            validate(partial::keyStoreFile),
+            validate(partial::keyStorePasswordFile),
+            validate(partial::trustStoreFile),
+            validate(partial::trustStorePasswordFile)
+    )
 
-    private fun createSecurityKeys(keyStorePath: Path,
-                                   keyStorePasswordPath: Path,
-                                   trustStorePath: Path,
-                                   trustStorePasswordPath: Path) =
-            ImmutableSecurityKeys.builder()
-                    .keyStore(ImmutableSecurityKeysStore.of(keyStorePath))
-                    .keyStorePassword(Passwords.fromPath(keyStorePasswordPath))
-                    .trustStore(ImmutableSecurityKeysStore.of(trustStorePath))
-                    .trustStorePassword(Passwords.fromPath(trustStorePasswordPath))
-                    .build()
+    private fun <A> validate(property: ConfigProperty<A>) =
+            Validated.fromOption(property.get(), { "- missing property: ${property.name}\n" })
 
-    private fun validatedCollectorConfig(partial: PartialConfiguration) =
-            partial.mapBinding { config ->
-                CollectorConfiguration(
-                        config.streamPublishers.bind().map { Route(it.name(), it) }
-                )
-            }
+    private fun <A> validationMessageFrom(invalidFields: List<Validated<String, A>>): String =
+            invalidFields.map { it as Invalid }
+                    .map { it.e }
+                    .fold("", String::plus)
+                    .let { "Some required configuration properties are missing: \n$it" }
 
     companion object {
-        val DEFAULT_LOG_LEVEL = LogLevel.INFO
         private val logger = Logger(ConfigurationValidator::class)
     }
 }
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/partial_configuration.kt
index 51f6a665..c8162104 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/partial_configuration.kt
@@ -21,9 +21,16 @@ package org.onap.dcae.collectors.veshv.config.impl
 
 import arrow.core.None
 import arrow.core.Option
+import arrow.core.Some
+import arrow.core.getOrElse
 import com.google.gson.annotations.SerializedName
+import org.onap.dcae.collectors.veshv.config.api.model.ValidationException
+import org.onap.dcae.collectors.veshv.utils.arrow.flatFold
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
+import kotlin.reflect.KProperty0
+
+internal typealias ConfigProperty<A> = KProperty0<Option<A>>
 
 /**
  * @author Pawel Biniek <pawel.biniek@nokia.com>
@@ -56,4 +63,29 @@ internal data class PartialConfiguration(
 
         @Transient
         var streamPublishers: Option<List<KafkaSink>> = None
-)
+) {
+    fun unsafeAsValidated() = ValidatedPartialConfiguration(
+            listenPort = getOrThrowValidationException(::listenPort),
+            idleTimeoutSec = getOrThrowValidationException(::idleTimeoutSec),
+            cbsConfiguration = ValidatedCbsConfiguration(
+                    firstRequestDelaySec = getOrThrowValidationException(::firstRequestDelaySec),
+                    requestIntervalSec = getOrThrowValidationException(::requestIntervalSec)
+            ),
+            streamPublishers = getOrThrowValidationException(::streamPublishers),
+            securityConfiguration = sslDisable.flatFold({ forceValidatedSecurityPaths() }, { None }),
+            logLevel = logLevel
+    )
+
+    private fun forceValidatedSecurityPaths() =
+            Some(ValidatedSecurityPaths(
+                    keyStoreFile = getOrThrowValidationException(::keyStoreFile),
+                    keyStorePasswordFile = getOrThrowValidationException(::keyStorePasswordFile),
+                    trustStoreFile = getOrThrowValidationException(::trustStoreFile),
+                    trustStorePasswordFile = getOrThrowValidationException(::trustStorePasswordFile)
+            ))
+}
+
+internal fun <A> getOrThrowValidationException(property: ConfigProperty<A>) =
+        property().getOrElse {
+            throw ValidationException("Field `${property.name}` was not validated and is missing in configuration")
+        }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/validated_configuration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/validated_configuration.kt
new file mode 100644
index 00000000..a230bfc0
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/validated_configuration.kt
@@ -0,0 +1,46 @@
+/*
+ * ============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.Option
+import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
+
+
+internal data class ValidatedPartialConfiguration(
+        val listenPort: Int,
+        val idleTimeoutSec: Long,
+        val cbsConfiguration: ValidatedCbsConfiguration,
+        val securityConfiguration: Option<ValidatedSecurityPaths>,
+        val logLevel: Option<LogLevel>,
+        val streamPublishers: List<KafkaSink>
+)
+
+internal data class ValidatedCbsConfiguration(
+        val firstRequestDelaySec: Long,
+        val requestIntervalSec: Long
+)
+
+internal data class ValidatedSecurityPaths(
+        val keyStoreFile: String,
+        val keyStorePasswordFile: String,
+        val trustStoreFile: String,
+        val trustStorePasswordFile: String
+)