Use SDK/SSL in HV-VES

Issue-ID: DCAEGEN2-1226
Change-Id: I7cfc09001f7315c1b6f4fcf150ad631630c810ef
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

1 files changed, 74 insertions, 65 deletions

diff --git a/sources/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgVesHvConfiguration.kt b/sources/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgVesHvConfiguration.kt
index c97486c6..56bb1d84 100644
--- a/sources/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgVesHvConfiguration.kt
+++ b/sources/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgVesHvConfiguration.kt
@@ -30,7 +30,9 @@ import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
 import org.onap.dcae.collectors.veshv.model.ConfigurationProviderParams
 import org.onap.dcae.collectors.veshv.model.KafkaConfiguration
 import org.onap.dcae.collectors.veshv.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.model.ServiceContext
 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.commandline.*
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.CONSUL_CONFIG_URL
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.KAFKA_SERVERS
@@ -55,80 +57,87 @@ import java.time.Duration
 
 internal class ArgVesHvConfiguration : ArgBasedConfiguration<ServerConfiguration>(DefaultParser()) {
     override val cmdLineOptionsList = listOf(
-        KAFKA_SERVERS,
-        HEALTH_CHECK_API_PORT,
-        LISTEN_PORT,
-        CONSUL_CONFIG_URL,
-        CONSUL_FIRST_REQUEST_DELAY,
-        CONSUL_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
+            KAFKA_SERVERS,
+            HEALTH_CHECK_API_PORT,
+            LISTEN_PORT,
+            CONSUL_CONFIG_URL,
+            CONSUL_FIRST_REQUEST_DELAY,
+            CONSUL_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).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()
+            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(ServiceContext::mdc) {
+                                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 configUrl = cmdLine.stringValue(CONSUL_CONFIG_URL).bind()
-            val firstRequestDelay = cmdLine.longValue(
-                CONSUL_FIRST_REQUEST_DELAY,
-                DefaultValues.CONSUL_FIRST_REQUEST_DELAY
-            )
-            val requestInterval = cmdLine.longValue(
-                CONSUL_REQUEST_INTERVAL,
-                DefaultValues.CONSUL_REQUEST_INTERVAL
-            )
-            ConfigurationProviderParams(
-                configUrl,
-                Duration.ofSeconds(firstRequestDelay),
-                Duration.ofSeconds(requestInterval)
-            )
-        }.fix()
+            Option.monad().binding {
+                val configUrl = cmdLine.stringValue(CONSUL_CONFIG_URL).bind()
+                val firstRequestDelay = cmdLine.longValue(
+                        CONSUL_FIRST_REQUEST_DELAY,
+                        DefaultValues.CONSUL_FIRST_REQUEST_DELAY
+                )
+                val requestInterval = cmdLine.longValue(
+                        CONSUL_REQUEST_INTERVAL,
+                        DefaultValues.CONSUL_REQUEST_INTERVAL
+                )
+                ConfigurationProviderParams(
+                        configUrl,
+                        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})"
+            .getOrElse {
+                logger.warn {
+                    "Failed to parse $logLevel as $LOG_LEVEL command line. " +
+                            "Using default log level (${DefaultValues.LOG_LEVEL})"
+                }
+                LogLevel.valueOf(DefaultValues.LOG_LEVEL)
             }
-            LogLevel.valueOf(DefaultValues.LOG_LEVEL)
-        }
 
 
     internal object DefaultValues {