Implement second part of dynamic DMaaP config

VESCollector app can now fetch CBS configuration and
rebuilt the part regarding sending events dynamically,
without restarting application.
Application will still be restarted by a .sh script,
if there were changes regarding collector.properties file.
The decision of whether dynamic configuration should be
triggered is now based on existence of env vars
CONSUL_HOST, CONFIG_BINDING_SERVICE, HOSTNAME,
not as previously on CBSPOLLTIME.
Frequency at which the config check should happen is
now exposed via property from collector.properties

Change-Id: I98ff160fa51d08d84a23c716d90ceaacbe17ada6
Signed-off-by: PawelSzalapski <pawel.szalapski@nokia.com>
Issue-ID: DCAEGEN2-519

16 files changed, 713 insertions, 340 deletions

diff --git a/src/main/java/org/onap/dcae/ApplicationSettings.java b/src/main/java/org/onap/dcae/ApplicationSettings.java
index 9063faa4..e4621849 100644
--- a/src/main/java/org/onap/dcae/ApplicationSettings.java
+++ b/src/main/java/org/onap/dcae/ApplicationSettings.java
@@ -34,7 +34,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Nullable;
-import java.io.File;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Base64;
 
@@ -48,9 +48,8 @@ import static java.util.Arrays.stream;
 public class ApplicationSettings {
 
     private static final Logger inlog = LoggerFactory.getLogger(ApplicationSettings.class);
-    private static final String COLLECTOR_PROPERTIES = "etc/collector.properties";
-
     private final String appInvocationDir;
+    private final String configurationFileLocation;
     private final PropertiesConfiguration properties = new PropertiesConfiguration();
 
     public ApplicationSettings(String[] args, Function1<String[], Map<String, String>> argsParser) {
@@ -61,20 +60,14 @@ public class ApplicationSettings {
         this.appInvocationDir = appInvocationDir;
         properties.setDelimiterParsingDisabled(true);
         Map<String, String> parsedArgs = argsParser.apply(args);
-        loadProperties(Paths.get(new File(COLLECTOR_PROPERTIES).getAbsolutePath()).toString());
-        loadCommandLineProperties(parsedArgs);
+        configurationFileLocation = findOutConfigurationFileLocation(parsedArgs);
+        loadPropertiesFromFile();
         parsedArgs.filterKeys(k -> !k.equals("c")).forEach(this::updateProperty);
     }
-    private void loadCommandLineProperties(Map<String, String> parsedArgs) {
-        parsedArgs.get("c").forEach(e -> {
-            properties.clear();
-            loadProperties(e);
-        });
-    }
 
-    private void loadProperties(String property) {
+    private void loadPropertiesFromFile() {
         try {
-            properties.load(property);
+            properties.load(configurationFileLocation);
         } catch (ConfigurationException ex) {
             inlog.error("Cannot load properties cause:", ex);
             throw new RuntimeException(ex);
@@ -90,6 +83,14 @@ public class ApplicationSettings {
                 .toMap(t -> t.split(",")[0].trim(), t -> new String(Base64.getDecoder().decode(t.split(",")[1])).trim());
     }
 
+    private String findOutConfigurationFileLocation(Map<String, String> parsedArgs) {
+        return prependWithUserDirOnRelative(parsedArgs.get("c").getOrElse("etc/collector.properties"));
+    }
+
+    public Path configurationFileLocation() {
+        return Paths.get(configurationFileLocation);
+    }
+
     public int maximumAllowedQueuedEvents() {
         return properties.getInt("collector.inputQueue.maxPending", 1024 * 4);
     }
@@ -115,6 +116,10 @@ public class ApplicationSettings {
         return properties.getInt("collector.service.secure.port", 8443);
     }
 
+    public int configurationUpdateFrequency() {
+        return properties.getInt("collector.dynamic.config.update.frequency", 5);
+    }
+
     public boolean httpsEnabled() {
         return httpsPort() > 0;
     }
@@ -139,7 +144,7 @@ public class ApplicationSettings {
         return properties.getString("exceptionConfig", null);
     }
 
-    public String cambriaConfigurationFileLocation() {
+    public String dMaaPConfigurationFileLocation() {
         return prependWithUserDirOnRelative(properties.getString("collector.dmaapfile", "etc/DmaapConfig.json"));
     }
 
diff --git a/src/main/java/org/onap/dcae/VesApplication.java b/src/main/java/org/onap/dcae/VesApplication.java
index 86b8ccb0..7eea0eb0 100644
--- a/src/main/java/org/onap/dcae/VesApplication.java
+++ b/src/main/java/org/onap/dcae/VesApplication.java
@@ -26,9 +26,11 @@ import org.onap.dcae.commonFunction.EventProcessor;
 import org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser;
 import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
 import org.onap.dcae.commonFunction.event.publishing.PublisherConfig;
+import org.onap.dcae.controller.ConfigLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.Banner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -38,9 +40,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Lazy;
 
 import java.nio.file.Paths;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.*;
 
 @SpringBootApplication
 @EnableAutoConfiguration(exclude = {GsonAutoConfiguration.class, SecurityAutoConfiguration.class})
@@ -61,21 +61,39 @@ public class VesApplication {
 
         fProcessingInputQueue = new LinkedBlockingQueue<>(properties.maximumAllowedQueuedEvents());
 
-        app.setAddCommandLineProperties(true);
-        app.run();
-
+        EventPublisher publisher = EventPublisher.createPublisher(oplog,
+                DMaaPConfigurationParser
+                        .parseToDomainMapping(Paths.get(properties.dMaaPConfigurationFileLocation()))
+                        .get());
+        spawnDynamicConfigUpdateThread(publisher, properties);
         EventProcessor ep = new EventProcessor(EventPublisher.createPublisher(oplog, getDmapConfig()), properties);
 
         ExecutorService executor = Executors.newFixedThreadPool(MAX_THREADS);
         for (int i = 0; i < MAX_THREADS; ++i) {
             executor.execute(ep);
         }
+
+        app.setBannerMode(Banner.Mode.OFF);
+        app.setAddCommandLineProperties(true);
+        app.run();
     }
 
+    private static void spawnDynamicConfigUpdateThread(EventPublisher eventPublisher, ApplicationSettings properties) {
+        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
+        ConfigLoader configLoader = ConfigLoader
+                .create(eventPublisher::reconfigure,
+                        Paths.get(properties.dMaaPConfigurationFileLocation()),
+                        properties.configurationFileLocation());
+        scheduledThreadPoolExecutor
+                .scheduleAtFixedRate(() -> configLoader.updateConfig(),
+                        properties.configurationUpdateFrequency(),
+                        properties.configurationUpdateFrequency(),
+                        TimeUnit.MINUTES);
+    }
 
     private static Map<String, PublisherConfig> getDmapConfig() {
         return DMaaPConfigurationParser.
-                parseToDomainMapping(Paths.get(properties.cambriaConfigurationFileLocation())).get();
+                parseToDomainMapping(Paths.get(properties.dMaaPConfigurationFileLocation())).get();
     }
 
     @Bean
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
index 179e8826..91db5172 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
@@ -29,6 +29,8 @@ import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
 
+import org.json.JSONObject;
+
 import static io.vavr.API.*;
 import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
 import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
@@ -45,6 +47,11 @@ public final class DMaaPConfigurationParser {
                 .flatMap(DMaaPConfigurationParser::toConfigMap);
     }
 
+    public static Try<Map<String, PublisherConfig>> parseToDomainMapping(JSONObject config) {
+        return toJSON(config.toString())
+            .flatMap(DMaaPConfigurationParser::toConfigMap);
+    }
+
     private static Try<String> readFromFile(Path configLocation) {
         return Try(() -> new String(Files.readAllBytes(configLocation)))
                 .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation)));
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
index 489fcbf0..4f672715 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
@@ -33,7 +33,6 @@ import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
  */
 final class DMaaPPublishersBuilder {
 
-    @SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do")
     static Try<CambriaBatchingPublisher> buildPublisher(PublisherConfig config) {
         return Try(() -> builder(config).build())
                 .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config)));
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
index 78f34ff4..7d535a21 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
@@ -21,13 +21,18 @@ package org.onap.dcae.commonFunction.event.publishing;
 
 import io.vavr.API;
 import io.vavr.API.Match.Case;
+import io.vavr.Function0;
+import io.vavr.Function1;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import org.slf4j.Logger;
 
 import static io.vavr.API.$;
 
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  */
-final class VavrUtils {
+public final class VavrUtils {
 
     private VavrUtils() {
         // utils aggregator
@@ -36,7 +41,7 @@ final class VavrUtils {
     /**
      * Shortcut for 'string interpolation'
      */
-    static String f(String msg, Object... args) {
+    public static String f(String msg, Object... args) {
         return String.format(msg, args);
     }
 
@@ -44,8 +49,17 @@ final class VavrUtils {
      * Wrap failure with a more descriptive message of what has failed and chain original cause. Used to provide a
      * context for errors instead of raw exception.
      */
-    static Case<Throwable, Throwable> enhanceError(String msg) {
+    public static Case<Throwable, Throwable> enhanceError(String msg) {
         return API.Case($(), e -> new RuntimeException(msg, e));
     }
 
+    public static Case<Throwable, Throwable> enhanceError(String pattern, Object... arguments) {
+        return API.Case($(), e -> new RuntimeException(f(pattern, arguments), e));
+    }
+
+    public static Consumer<Throwable> logError(Logger withLogger) {
+        return e -> withLogger.error(e.getMessage(), e);
+    }
+
+
 }
diff --git a/src/main/java/org/onap/dcae/controller/ConfigFilesFacade.java b/src/main/java/org/onap/dcae/controller/ConfigFilesFacade.java
new file mode 100644
index 00000000..42155eda
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/ConfigFilesFacade.java
@@ -0,0 +1,122 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static io.vavr.API.Try;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.logError;
+import static org.onap.dcae.controller.Conversions.toList;
+
+import io.vavr.CheckedRunnable;
+import io.vavr.Tuple2;
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.io.FileNotFoundException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class ConfigFilesFacade {
+
+    private static Logger log = LoggerFactory.getLogger(ConfigFilesFacade.class);
+
+    private final Path dMaaPConfigPath;
+    private final Path propertiesPath;
+
+    public ConfigFilesFacade(Path dMaaPConfigPath, Path propertiesPath) {
+        this.dMaaPConfigPath = dMaaPConfigPath;
+        this.propertiesPath = propertiesPath;
+    }
+
+    Try<Map<String, String>> readCollectorProperties() {
+        log.info(f("Reading collector properties from path: '%s'", propertiesPath));
+        return Try(() -> readProperties())
+            .map(prop -> toList(prop.getKeys()).toMap(k -> k, k -> (String) prop.getProperty(k)))
+            .mapFailure(enhanceError("Unable to read properties configuration from path '%s'", propertiesPath))
+            .onFailure(logError(log))
+            .peek(props -> log.info(f("Read following collector properties: '%s'", props)));
+    }
+
+    Try<JSONObject> readDMaaPConfiguration() {
+        log.info(f("Reading DMaaP configuration from file: '%s'", dMaaPConfigPath));
+        return readFile(dMaaPConfigPath)
+            .recover(FileNotFoundException.class, __ -> "{}")
+            .mapFailure(enhanceError("Unable to read DMaaP configuration from file '%s'", dMaaPConfigPath))
+            .flatMap(Conversions::toJson)
+            .onFailure(logError(log))
+            .peek(props -> log.info(f("Read following DMaaP properties: '%s'", props)));
+    }
+
+    Try<Void> writeDMaaPConfiguration(JSONObject dMaaPConfiguration) {
+        log.info(f("Writing DMaaP configuration '%s' into file '%s'", dMaaPConfiguration, dMaaPConfigPath));
+        return writeFile(dMaaPConfigPath, indentConfiguration(dMaaPConfiguration.toString()))
+            .mapFailure(enhanceError("Could not save new DMaaP configuration to path '%s'", dMaaPConfigPath))
+            .onFailure(logError(log))
+            .peek(__ -> log.info("Written successfully"));
+    }
+
+
+    Try<Void> writeProperties(Map<String, String> properties) {
+        log.info(f("Writing properties configuration '%s' into file '%s'", properties, propertiesPath));
+        return Try.run(saveProperties(properties))
+            .mapFailure(enhanceError("Could not save properties to path '%s'", properties))
+            .onFailure(logError(log))
+            .peek(__ -> log.info("Written successfully"));
+    }
+
+    private Try<String> readFile(Path path) {
+        return Try(() -> new String(Files.readAllBytes(path), StandardCharsets.UTF_8))
+            .mapFailure(enhanceError("Could not read content from path: '%s'", path));
+    }
+
+    private Try<Void> writeFile(Path path, String content) {
+        return Try.run(() -> Files.write(path, content.getBytes()))
+            .mapFailure(enhanceError("Could not write content to path: '%s'", path));
+    }
+
+    private PropertiesConfiguration readProperties() throws ConfigurationException {
+        PropertiesConfiguration propertiesConfiguration = new PropertiesConfiguration();
+        propertiesConfiguration.load(propertiesPath.toFile());
+        return propertiesConfiguration;
+    }
+
+    private CheckedRunnable saveProperties(Map<String, String> properties) {
+        return () -> {
+            PropertiesConfiguration propertiesConfiguration = new PropertiesConfiguration(propertiesPath.toFile());
+            propertiesConfiguration.setEncoding(null);
+            for (Tuple2<String, String> property : properties) {
+                propertiesConfiguration.addProperty(property._1, property._2);
+            }
+            propertiesConfiguration.save();
+        };
+    }
+
+    private String indentConfiguration(String configuration) {
+        return new JSONObject(configuration).toString(4);
+    }
+
+}
diff --git a/src/main/java/org/onap/dcae/controller/ConfigLoader.java b/src/main/java/org/onap/dcae/controller/ConfigLoader.java
new file mode 100644
index 00000000..fb807075
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/ConfigLoader.java
@@ -0,0 +1,128 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser.parseToDomainMapping;
+import static org.onap.dcae.controller.ConfigParsing.getDMaaPConfig;
+import static org.onap.dcae.controller.ConfigParsing.getProperties;
+import static org.onap.dcae.controller.EnvPropertiesReader.readEnvProps;
+
+import io.vavr.Function0;
+import io.vavr.Function1;
+import io.vavr.collection.HashMap;
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.nio.file.Path;
+import java.util.function.Consumer;
+import org.json.JSONObject;
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
+import org.onap.dcae.commonFunction.event.publishing.PublisherConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigLoader {
+
+    private static final String SKIP_MSG = "Skipping dynamic configuration update";
+    private static Logger log = LoggerFactory.getLogger(ConfigLoader.class);
+    private final Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer;
+    private final ConfigFilesFacade configFilesFacade;
+    private final Function1<EnvProps, Try<JSONObject>> configurationSource;
+    private final Function0<Map<String, String>> envVariablesSupplier;
+
+    ConfigLoader(Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer,
+                 ConfigFilesFacade configFilesFacade,
+                 Function1<EnvProps, Try<JSONObject>> configurationSource,
+                 Function0<Map<String, String>> envVariablesSupplier) {
+        this.eventPublisherReconfigurer = eventPublisherReconfigurer;
+        this.configFilesFacade = configFilesFacade;
+        this.configurationSource = configurationSource;
+        this.envVariablesSupplier = envVariablesSupplier;
+    }
+
+    public static ConfigLoader create(Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer,
+                                      Path dMaaPConfigFile, Path propertiesConfigFile) {
+        return new ConfigLoader(eventPublisherReconfigurer,
+            new ConfigFilesFacade(dMaaPConfigFile, propertiesConfigFile),
+            ConfigSource::getAppConfig,
+            () -> HashMap.ofAll(System.getenv()));
+    }
+
+    public void updateConfig() {
+        log.info("Trying to dynamically update config from Config Binding Service");
+        readEnvProps(envVariablesSupplier.get())
+            .onEmpty(() -> log.warn(SKIP_MSG))
+            .forEach(props -> updateConfig(props));
+    }
+
+    private void updateConfig(EnvProps props) {
+        configurationSource.apply(props)
+            .onFailure(logSkip())
+            .onSuccess(newConf -> {
+                    updateConfigurationProperties(newConf);
+                    updateDMaaPProperties(newConf);
+                }
+            );
+    }
+
+    private void updateDMaaPProperties(JSONObject newConf) {
+        configFilesFacade.readDMaaPConfiguration()
+            .onFailure(logSkip())
+            .onSuccess(oldDMaaPConf -> getDMaaPConfig(newConf)
+                .onEmpty(() -> log.warn(SKIP_MSG))
+                .forEach(newDMaaPConf -> compareAndOverwriteDMaaPConfig(oldDMaaPConf, newDMaaPConf)));
+    }
+
+
+    private void updateConfigurationProperties(JSONObject newConf) {
+        configFilesFacade.readCollectorProperties()
+            .onFailure(logSkip())
+            .onSuccess(oldProps -> compareAndOverwritePropertiesConfig(newConf, oldProps));
+    }
+
+    private void compareAndOverwritePropertiesConfig(JSONObject newConf, Map<String, String> oldProps) {
+        Map<String, String> newProperties = getProperties(newConf);
+        if (!oldProps.equals(newProperties)) {
+            configFilesFacade.writeProperties(newProperties)
+                .onSuccess(__ -> log.info("New properties configuration written to file"))
+                .onFailure(logSkip());
+        } else {
+            log.info("Collector properties from CBS are the same as currently used ones. " + SKIP_MSG);
+        }
+    }
+
+    private void compareAndOverwriteDMaaPConfig(JSONObject oldDMaaPConf, JSONObject newDMaaPConf) {
+        if (!oldDMaaPConf.toString().equals(newDMaaPConf.toString())) {
+            parseToDomainMapping(newDMaaPConf)
+                .onFailure(exc -> log.error(SKIP_MSG, exc))
+                .onSuccess(eventPublisherReconfigurer)
+                .onSuccess(parsedConfig ->
+                    configFilesFacade.writeDMaaPConfiguration(newDMaaPConf)
+                        .onFailure(logSkip())
+                        .onSuccess(__ -> log.info("New dMaaP configuration written to file")));
+        } else {
+            log.info("DMaaP config from CBS is the same as currently used one. " + SKIP_MSG);
+        }
+    }
+
+    private Consumer<Throwable> logSkip() {
+        return __ -> log.error(SKIP_MSG);
+    }
+}
diff --git a/src/main/java/org/onap/dcae/controller/ConfigParsing.java b/src/main/java/org/onap/dcae/controller/ConfigParsing.java
new file mode 100644
index 00000000..2ee0c918
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/ConfigParsing.java
@@ -0,0 +1,58 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static io.vavr.API.Try;
+import static io.vavr.API.Tuple;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+import static org.onap.dcae.controller.Conversions.toList;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+interface ConfigParsing {
+
+    Logger log = LoggerFactory.getLogger(ConfigParsing.class);
+
+    static Option<JSONObject> getDMaaPConfig(JSONObject configuration) {
+        log.info(f("Getting DMaaP configuration from app configuration: '%s'", configuration));
+        return toList(configuration.toMap().entrySet().iterator())
+            .filter(t -> t.getKey().startsWith("streams_publishes"))
+            .headOption()
+            .flatMap(e -> Try(() -> configuration.getJSONObject(e.getKey())).toOption())
+            .onEmpty(() -> log.warn(f("App configuration '%s' is missing DMaaP configuration ('streams_publishes' key) "
+                + "or DMaaP configuration is not a valid json document", configuration)))
+            .peek(dMaaPConf -> log.info(f("Found following DMaaP configuration: '%s'", dMaaPConf)));
+    }
+
+    static Map<String, String> getProperties(JSONObject configuration) {
+        log.info(f("Getting properties configuration from app configuration: '%s'", configuration));
+        Map<String, String> confEntries = toList(configuration.toMap().entrySet().iterator())
+            .toMap(e -> Tuple(e.getKey(), String.valueOf(e.getValue())))
+            .filterKeys(e -> !e.startsWith("streams_publishes"));
+        log.info(f("Found following app properties: '%s'", confEntries));
+        return confEntries;
+    }
+
+}
diff --git a/src/main/java/org/onap/dcae/controller/ConfigSource.java b/src/main/java/org/onap/dcae/controller/ConfigSource.java
new file mode 100644
index 00000000..7e6a9fc8
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/ConfigSource.java
@@ -0,0 +1,88 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static io.vavr.API.Try;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+import static org.onap.dcae.controller.Conversions.toJson;
+import static org.onap.dcae.controller.Conversions.toJsonArray;
+
+import com.mashape.unirest.http.Unirest;
+import io.vavr.control.Try;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class ConfigSource {
+
+    private static final Logger log = LoggerFactory.getLogger(ConfigSource.class);
+
+    static Try<JSONObject> getAppConfig(EnvProps envProps) {
+        log.info("Fetching app configuration from CBS");
+        return callConsulForCBSConfiguration(envProps)
+            .peek(strBody -> log.info(f("Received following CBS configuration from Consul '%s'", strBody)))
+            .flatMap(strBody -> toJsonArray(strBody))
+            .flatMap(json -> withdrawCatalog(json))
+            .flatMap(json -> constructFullCBSUrl(json))
+            .flatMap(cbsUrl -> callCBSForAppConfig(envProps, cbsUrl))
+            .flatMap(strBody -> toJson(strBody))
+            .peek(jsonNode -> log.info(f("Received app configuration: '%s'", jsonNode)))
+            .onFailure(exc -> log.error("Could not fetch application config", exc));
+    }
+
+    private static Try<String> callConsulForCBSConfiguration(EnvProps envProps) {
+        return executeGet(envProps.consulHost + ":" + envProps.consulPort + "/v1/catalog/service/" + envProps.cbsName)
+            .mapFailure(enhanceError("Unable to retrieve CBS configuration from Consul"));
+    }
+
+    private static Try<String> constructFullCBSUrl(JSONObject json) {
+        return Try(() -> json.get("ServiceAddress").toString() + ":" + json.get("ServicePort").toString())
+            .mapFailure(enhanceError("ServiceAddress / ServicePort missing from CBS conf: '%s'", json));
+    }
+
+    private static Try<JSONObject> withdrawCatalog(JSONArray json) {
+        return Try(() -> new JSONObject(json.get(0).toString()))
+            .mapFailure(enhanceError("CBS response '%s' is in invalid format,"
+                + " most probably is it not a list of configuration objects", json));
+    }
+
+    private static Try<String> callCBSForAppConfig(EnvProps envProps, String cbsUrl) {
+        log.info("Calling CBS for application config");
+        return executeGet(cbsUrl + "/service_component/" + envProps.appName)
+            .mapFailure(enhanceError("Unable to fetch configuration from CBS"));
+    }
+
+
+    private static Try<String> executeGet(String url) {
+        log.info(f("Calling HTTP GET on url: '%s'", url));
+        return Try(() -> Unirest.get(url).asString())
+            .mapFailure(enhanceError("Http call (GET '%s') failed.", url))
+            .filter(
+                res -> res.getStatus() == 200,
+                res -> new RuntimeException(f("HTTP call (GET '%s') failed with status %s and body '%s'",
+                    url, res.getStatus(), res.getBody())))
+            .map(res -> res.getBody())
+            .peek(body -> log.info(f("HTTP GET on '%s' returned body '%s'", url, body)));
+    }
+
+}
diff --git a/src/main/java/org/onap/dcae/controller/Conversions.java b/src/main/java/org/onap/dcae/controller/Conversions.java
new file mode 100644
index 00000000..09a9a43c
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/Conversions.java
@@ -0,0 +1,53 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+
+import io.vavr.API;
+import io.vavr.collection.List;
+import io.vavr.control.Try;
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.StreamSupport;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+interface Conversions {
+
+    static Try<JSONObject> toJson(String strBody) {
+        return API.Try(() -> new JSONObject(strBody))
+            .mapFailure(enhanceError("Value '%s' is not a valid JSON document", strBody));
+    }
+
+    static Try<JSONArray> toJsonArray(String strBody) {
+        return API.Try(() -> new JSONArray(strBody))
+            .mapFailure(enhanceError("Value '%s' is not a valid JSON array", strBody));
+    }
+
+    static <T> List<T> toList(Iterator<T> iterator) {
+        return List.ofAll(StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false));
+    }
+
+}
diff --git a/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java b/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java
new file mode 100644
index 00000000..23bcbda8
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java
@@ -0,0 +1,77 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import static io.vavr.API.List;
+import static io.vavr.API.Try;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class EnvPropertiesReader {
+
+    private final static Logger log = LoggerFactory.getLogger(EnvPropertiesReader.class);
+
+    static Option<EnvProps> readEnvProps(Map<String, String> environmentVariables) {
+        log.info("Loading necessary environment variables for dynamic configuration update");
+        int consulPort = getConsulPort(environmentVariables);
+        Option<String> consulHost = getConsulHost(environmentVariables);
+        Option<String> cbsServiceName = getCBSName(environmentVariables);
+        Option<String> vesCollectorAppName = getAppName(environmentVariables);
+        return Option.sequence(List(consulHost, cbsServiceName, vesCollectorAppName))
+            .map(e -> new EnvProps(e.get(0), consulPort, e.get(1), e.get(2)))
+            .onEmpty(() -> log.warn("Some required environment variables are missing"))
+            .peek(props -> log.info(f("Discovered following environment variables: '%s'", props)));
+    }
+
+    private static Option<String> getAppName(Map<String, String> environmentVariables) {
+        return environmentVariables.get("HOSTNAME")
+            .orElse(environmentVariables.get("SERVICE_NAME"))
+            .onEmpty(() -> log.warn("App service name (as registered in CBS) (env var: 'HOSTNAME' / 'SERVICE_NAME') "
+                + "is missing error environment variables."));
+    }
+
+    private static Option<String> getCBSName(Map<String, String> environmentVariables) {
+        return environmentVariables.get("CONFIG_BINDING_SERVICE")
+            .onEmpty(() -> log.warn("Name of CBS Service (as registered in Consul) (env var: 'CONFIG_BINDING_SERVICE') "
+                + "is missing from environment variables."));
+    }
+
+    private static Integer getConsulPort(Map<String, String> environmentVariables) {
+        return environmentVariables.get("CONSUL_PORT")
+            .flatMap(str -> Try(() -> Integer.valueOf(str))
+                .onFailure(exc -> log.warn("Consul port is not an integer value", exc))
+                .toOption())
+            .onEmpty(() -> log.warn("Consul port (env var: 'CONSUL_PORT') is missing from environment variables. "
+                + "Using default value of 8500"))
+            .getOrElse(8500);
+    }
+
+    private static Option<String> getConsulHost(Map<String, String> environmentVariables) {
+        return environmentVariables.get("CONSUL_HOST")
+            .onEmpty(() -> log.warn("Consul host (env var: 'CONSUL_HOST') (without port) "
+                + "is missing from environment variables."));
+    }
+
+}
diff --git a/src/main/java/org/onap/dcae/controller/EnvProps.java b/src/main/java/org/onap/dcae/controller/EnvProps.java
new file mode 100644
index 00000000..2ee41cc6
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/EnvProps.java
@@ -0,0 +1,70 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import java.util.Objects;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+final class EnvProps {
+
+    final String consulHost;
+    final int consulPort;
+    final String cbsName;
+    final String appName;
+
+    EnvProps(String consulHost, int consulPort, String cbsName, String appName) {
+        this.consulHost = consulHost;
+        this.consulPort = consulPort;
+        this.cbsName = cbsName;
+        this.appName = appName;
+    }
+
+    @Override
+    public String toString() {
+        return "EnvProps{" +
+            "consulHost='" + consulHost + '\'' +
+            ", consulPort=" + consulPort +
+            ", cbsName='" + cbsName + '\'' +
+            ", appName='" + appName + '\'' +
+            '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        EnvProps envProps = (EnvProps) o;
+        return consulPort == envProps.consulPort &&
+            Objects.equals(consulHost, envProps.consulHost) &&
+            Objects.equals(cbsName, envProps.cbsName) &&
+            Objects.equals(appName, envProps.appName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(consulHost, consulPort, cbsName, appName);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java b/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java
deleted file mode 100644
index ed42a5a4..00000000
--- a/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java
+++ /dev/null
@@ -1,186 +0,0 @@
-/*-

- * ============LICENSE_START=======================================================

- * PROJECT

- * ================================================================================

- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.

- * ================================================================================

- * 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.controller;

-

-import com.fasterxml.jackson.databind.JsonNode;

-import com.fasterxml.jackson.databind.ObjectMapper;

-import org.json.JSONArray;

-import org.json.JSONObject;

-import org.json.JSONTokener;

-import org.slf4j.Logger;

-import org.slf4j.LoggerFactory;

-

-import java.io.*;

-import java.util.Map;

-

-public class FetchDynamicConfig {

-

-    private static final Logger log = LoggerFactory.getLogger(FetchDynamicConfig.class);

-

-    public static String configFile = "/opt/app/KV-Configuration.json";

-    public static String retString;

-    public static String retCBSString;

-    private static String url;

-    private static Map<String, String> env;

-

-    public FetchDynamicConfig() {

-    }

-

-    public static void main(String[] args) {

-        Boolean areEqual;

-        // Call consul api and identify the CBS Service address and port

-        getconsul();

-        // Construct and invoke CBS API to get application Configuration

-        getCBS();

-        // Verify if data has changed

-        areEqual = verifyConfigChange();

-        // If new config then write data returned into configFile for

-        // LoadDynamicConfig process

-        if (!areEqual) {

-            FetchDynamicConfig fc = new FetchDynamicConfig();

-            fc.writefile(retCBSString);

-        } else {

-            log.info("New config pull results identical -  " + configFile + " NOT refreshed");

-        }

-    }

-

-    private static void getconsul() {

-

-        env = System.getenv();

-        for (Map.Entry<String, String> entry : env.entrySet()) {

-            log.info(entry.getKey() + ":" + entry.getValue());

-        }

-

-        if (env.containsKey("CONSUL_HOST") && env.containsKey("CONFIG_BINDING_SERVICE")) {

-            // && env.containsKey("HOSTNAME")) {

-            log.info(">>>Dynamic configuration to be fetched from ConfigBindingService");

-            url = env.get("CONSUL_HOST") + ":8500/v1/catalog/service/" + env.get("CONFIG_BINDING_SERVICE");

-

-            retString = executecurl(url);

-

-        } else {

-            log.info(">>>Static configuration to be used");

-        }

-

-    }

-

-    public static boolean verifyConfigChange() {

-

-        boolean areEqual = false;

-        // Read current data

-        try {

-            File f = new File(configFile);

-            if (f.exists() && !f.isDirectory()) {

-

-                String jsonData = LoadDynamicConfig.readFile(configFile);

-                JSONObject jsonObject = new JSONObject(jsonData);

-

-                ObjectMapper mapper = new ObjectMapper();

-

-                JsonNode tree1 = mapper.readTree(jsonObject.toString());

-                JsonNode tree2 = mapper.readTree(retCBSString);

-                areEqual = tree1.equals(tree2);

-                log.info("Comparison value:" + areEqual);

-            } else {

-                log.info("First time config file read: " + configFile);

-            }

-

-        } catch (IOException e) {

-            log.error("Comparison with new fetched data failed" + e.getMessage());

-

-        }

-

-        return areEqual;

-

-    }

-

-    public static void getCBS() {

-

-        env = System.getenv();

-        // consul return as array

-        JSONTokener temp = new JSONTokener(retString);

-        JSONObject cbsjobj = (JSONObject) new JSONArray(temp).get(0);

-

-        String urlPart1 = null;

-        if (cbsjobj.has("ServiceAddress") && cbsjobj.has("ServicePort")) {

-            urlPart1 = cbsjobj.getString("ServiceAddress") + ":" + cbsjobj.getInt("ServicePort");

-        }

-

-        log.info("CONFIG_BINDING_SERVICE DNS RESOLVED:" + urlPart1);

-

-        if (env.containsKey("HOSTNAME")) {

-            url = urlPart1 + "/service_component/" + env.get("HOSTNAME");

-            retCBSString = executecurl(url);

-        } else if (env.containsKey("SERVICE_NAME")) {

-            url = urlPart1 + "/service_component/" + env.get("SERVICE_NAME");

-            retCBSString = executecurl(url);

-        } else {

-            log.error("Service name environment variable - HOSTNAME/SERVICE_NAME not found within container ");

-        }

-

-    }

-

-    private static String executecurl(String url) {

-

-        String[] command = {"curl", "-v", url};

-        ProcessBuilder process = new ProcessBuilder(command);

-        Process p;

-        String result = null;

-        try {

-            p = process.start();

-            InputStreamReader ipr = new InputStreamReader(p.getInputStream());

-            BufferedReader reader = new BufferedReader(ipr);

-            StringBuilder builder = new StringBuilder();

-            String line;

-

-            while ((line = reader.readLine()) != null) {

-                builder.append(line);

-            }

-            result = builder.toString();

-            log.info(result);

-

-            reader.close();

-            ipr.close();

-        } catch (IOException e) {

-            log.error("error", e);

-            e.printStackTrace();

-        }

-        return result;

-

-    }

-

-    public void writefile(String retCBSString) {

-        log.info("URL to fetch configuration:" + url + " Return String:" + retCBSString);

-

-        String indentedretstring = (new JSONObject(retCBSString)).toString(4);

-

-        try (FileWriter file = new FileWriter(FetchDynamicConfig.configFile)) {

-            file.write(indentedretstring);

-

-            log.info("Successfully Copied JSON Object to file " + configFile);

-        } catch (IOException e) {

-            log.error("Error in writing configuration into file " + configFile + retString + e.getMessage());

-            e.printStackTrace();

-        }

-

-    }

-

-}

diff --git a/src/main/java/org/onap/dcae/controller/LoadDynamicConfig.java b/src/main/java/org/onap/dcae/controller/LoadDynamicConfig.java
deleted file mode 100644
index c1ab80c1..00000000
--- a/src/main/java/org/onap/dcae/controller/LoadDynamicConfig.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.controller;
-
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.PropertiesConfiguration;
-import org.json.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedReader;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.Map;
-
-public class LoadDynamicConfig {
-
-    private static final Logger log = LoggerFactory.getLogger(LoadDynamicConfig.class);
-
-    public String propFile = "collector.properties";
-    public String configFile = "/opt/app/KV-Configuration.json";
-    public String dMaaPOutputFile = "./etc/DmaapConfig.json";
-
-    public LoadDynamicConfig() {
-
-    }
-
-    public static void main(String[] args) {
-        Map<String, String> env = System.getenv();
-
-        // Check again to ensure new controller deployment related config
-        if (env.containsKey("CONSUL_HOST") && env.containsKey("CONFIG_BINDING_SERVICE")
-                && env.containsKey("HOSTNAME")) {
-
-            try {
-
-                LoadDynamicConfig lc = new LoadDynamicConfig();
-                String jsonData = readFile(lc.configFile);
-                JSONObject jsonObject = new JSONObject(jsonData);
-                lc.writeconfig(jsonObject);
-
-
-            } catch (Exception e) {
-                log.error(e.getLocalizedMessage(), e);
-                e.printStackTrace();
-
-            }
-
-        } else {
-            log.info(">>>Static configuration to be used");
-        }
-
-    }
-
-    public static String readFile(String filename) {
-        String result = "";
-        try (BufferedReader br = new BufferedReader(new FileReader(filename))) {
-            StringBuilder sb = new StringBuilder();
-            String line = br.readLine();
-            while (line != null) {
-                sb.append(line);
-                line = br.readLine();
-            }
-            result = sb.toString();
-        } catch (Exception e) {
-            log.error(e.getLocalizedMessage(), e);
-            e.printStackTrace();
-        }
-        return result;
-    }
-
-    public void writeconfig(JSONObject jsonObject) {
-
-        PropertiesConfiguration conf;
-        try {
-            conf = new PropertiesConfiguration(propFile);
-
-            conf.setEncoding(null);
-
-            // update properties based on consul dynamic configuration
-            Iterator<?> keys = jsonObject.keys();
-
-            while (keys.hasNext()) {
-                String key = (String) keys.next();
-                // check if any configuration is related to dmaap
-                // and write into dmaapconfig.json
-                if (key.startsWith("streams_publishes")) {
-                    // VESCollector only have publish streams
-                    try (FileWriter file = new FileWriter(dMaaPOutputFile)) {
-                        String indentedretstring = (new JSONObject(jsonObject.get(key).toString())).toString(4);
-                        file.write(indentedretstring);
-                        log.info("Successfully written JSON Object to DmaapConfig.json");
-                    } catch (IOException e) {
-                        log.info("Error in writing dmaap configuration into DmaapConfig.json", e);
-                    }
-                } else {
-                    conf.setProperty(key, jsonObject.get(key).toString());
-                }
-
-            }
-            conf.save();
-        } catch (ConfigurationException e) {
-            log.error(e.getLocalizedMessage(), e);
-            e.printStackTrace();
-        }
-    }
-
-}
diff --git a/src/main/java/org/onap/dcae/controller/PreAppStartupConfigUpdater.java b/src/main/java/org/onap/dcae/controller/PreAppStartupConfigUpdater.java
new file mode 100644
index 00000000..c5ee9d86
--- /dev/null
+++ b/src/main/java/org/onap/dcae/controller/PreAppStartupConfigUpdater.java
@@ -0,0 +1,49 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.controller;
+
+import io.vavr.collection.Map;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.function.Consumer;
+import org.onap.dcae.commonFunction.event.publishing.PublisherConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * On the first application launch, the configuration update thread that application spawns, has no chance to run yet
+ * and prepare initial application configuration. In this case, it needs to be fetched from outside of the application,
+ * so this is run from the .sh script.
+ * Later on, once application is already started it will take care of the configuration update itself
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public class PreAppStartupConfigUpdater {
+    private final static Logger log = LoggerFactory.getLogger(PreAppStartupConfigUpdater.class);
+
+    private static final Path DEFAULT_CONFIGURATION_FILE_PATH = Paths.get("etc/collector.properties");
+    private static final Path DEFAULT_DMAAP_FILE_PATH = Paths.get("etc/DmaapConfig.json");
+    private static final Consumer<Map<String, PublisherConfig>> NO_OP_CONSUMER = c -> { };
+
+    public static void main(String[] args) {
+        log.info("Running initial configuration update, before the application gets started.");
+        ConfigLoader.create(NO_OP_CONSUMER, DEFAULT_DMAAP_FILE_PATH, DEFAULT_CONFIGURATION_FILE_PATH)
+            .updateConfig();
+    }
+}
diff --git a/src/main/java/org/onap/dcae/restapi/VesRestController.java b/src/main/java/org/onap/dcae/restapi/VesRestController.java
index b7fc5f3b..92e8d004 100644
--- a/src/main/java/org/onap/dcae/restapi/VesRestController.java
+++ b/src/main/java/org/onap/dcae/restapi/VesRestController.java
@@ -23,7 +23,6 @@ package org.onap.dcae.restapi;
 
 import static java.util.Optional.ofNullable;
 import static java.util.stream.StreamSupport.stream;
-import static org.springframework.http.ResponseEntity.accepted;
 import static org.springframework.http.ResponseEntity.ok;
 
 import com.att.nsa.clock.SaClock;