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

31 files changed, 1616 insertions, 794 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;
diff --git a/src/main/scripts/VESConfigPoller.sh b/src/main/scripts/VESConfigPoller.sh
deleted file mode 100644
index 75c2b585..00000000
--- a/src/main/scripts/VESConfigPoller.sh
+++ /dev/null
@@ -1,125 +0,0 @@
-#!/bin/sh -x
-###
-# ============LICENSE_START=======================================================
-# PROJECT
-# ================================================================================
-# Copyright (C) 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=========================================================
-###
-# redirect stdout/stderr to a file
-#exec &> /opt/app/VESCollector/logs/console.txt
-
-usage() {
-        echo "VESConfigPoller.sh"
-}
-
-
-## Remove singel execution logic (loop 0)
-## On configupdate function, remove LoadDynamicConfig and invoke VESrestfulCollector stop/start
-
-BASEDIR=/opt/app/VESCollector/
-CONFIGFILENAME=/opt/app/KV-Configuration.json
-
-
-collector_configupdate() {
-
-        echo `date +"%Y%m%d.%H%M%S%3N"` - VESConfigPoller.sh:collector_configupdate
-        if [ -z "$CONSUL_HOST" ] || [ -z "$CONFIG_BINDING_SERVICE" ] || [ -z "$HOSTNAME" ]; then
-                echo "INFO: USING STANDARD CONTROLLER CONFIGURATION"
-        else
-            # move into base directory
-            cd $BASEDIR
-
-            CONFIG_FETCH=org.onap.dcae.controller.FetchDynamicConfig
-            $JAVA -cp "etc${PATHSEP}lib/*"  $CONFIG_FETCH $*
-            if [ $? -ne 0 ]; then
-                echo "ERROR: Failed to fetch dynamic configuration from consul into container $CONFIGFILENAME"
-            else
-               echo "INFO: Dynamic config fetched successfully"
-            fi
-                sleep 10s
-                FLAG=0
-
-            if [ -f $CONFIGFILENAME ]; then
-                if [[ $(find $CONFIGFILENAME -mmin -$CBSPOLLTIMER -print) ]]; then
-                        echo "File  $CONFIGFILENAME  is updated under $CBSPOLLTIMER minutes; Loader to be invoked"
-                        FLAG=1
-                else
-                        echo "File  $CONFIGFILENAME  NOT updated in last $CBSPOLLTIMER minutes; no configuration update!"
-                        FLAG=0
-                fi
-
-                if [ $FLAG -eq 1 ]; then
-                        echo "INFO: CONFIGFILE updated; triggering restart"
-                        /opt/app/VESCollector/bin/VESrestfulCollector.sh stop
-                        /opt/app/VESCollector/bin/VESrestfulCollector.sh start &
-                else
-                        echo "INFO: CONFIGFILE load skipped"
-                fi
-            else
-                echo "ERROR: Configuration file $CONFIGFILENAME missing"
-            fi
-        fi
-}
-
-
-
-if [ -z "$CBSPOLLTIMER" ]; then
-        echo "CBSPOLLTIMER not set; set this to polling frequency in minutes"
-        exit 1
-fi
-
-
-## Pre-setting
-
-# use JAVA_HOME if provided
-if [ -z "$JAVA_HOME" ]; then
-        echo "ERROR: JAVA_HOME not setup"
-        echo "Startup Aborted!!"
-        exit 1
-        #JAVA=java
-else
-        JAVA=$JAVA_HOME/bin/java
-fi
-
-
-
-# determine a path separator that works for this platform
-PATHSEP=":"
-case "$(uname -s)" in
-
-        Darwin)
-                ;;
-
-         Linux)
-                ;;
-
-         CYGWIN*|MINGW32*|MSYS*)
-                PATHSEP=";"
-                ;;
-
-        *)
-                ;;
-esac
-
-
-
-##Run in loop the config pull and check
-while true
-do
-        sleep $(echo $CBSPOLLTIMER)m
-        collector_configupdate | tee -a ${BASEDIR}/logs/console.txt
-done
-
diff --git a/src/main/scripts/VESrestfulCollector.sh b/src/main/scripts/VESrestfulCollector.sh
deleted file mode 100644
index 7f6d17cb..00000000
--- a/src/main/scripts/VESrestfulCollector.sh
+++ /dev/null
@@ -1,102 +0,0 @@
-#!/bin/bash

-

-###

-# ============LICENSE_START=======================================================

-# PROJECT

-# ================================================================================

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

-# Copyright (C) 2018 Nokia Networks 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=========================================================

-###

-source bin/logger.sh

-

-start() {

-    log "Starting application"

-    appPids=`pidof java`

-

-    if [ ! -z ${appPids} ]; then

-        logWarn "Tried to start an application, but it is already running on PID(s): ${appPids}. Startup aborted."

-        exit 1

-    fi

-

-    ${JAVA_HOME}/bin/java -cp "etc:lib/*" \

-      -Xms256m -Xmx512m \

-      -XX:ErrorFile=logs/java_error%p.log \

-      -XX:+HeapDumpOnOutOfMemoryError \

-      -Dhttps.protocols=TLSv1.1,TLSv1.2 \

-      org.onap.dcae.VesApplication $* & &>> logs/collector.log

-}

-

-stop() {

-    log "Stopping application"

-    appPids=`pidof java`

-

-    if [ ! -z ${appPids} ]; then

-        echo "Killing java PID(s): ${appPids}"

-        kill -9 ${appPids}

-        sleep 5

-        if [ ! $(pidof java) ]; then

-            log "Application stopped"

-        else

-            logWarn "Application did not stop after 5 seconds"

-        fi

-    else

-        logWarn "Tried to stop an application, but it was not running";

-    fi

-}

-

-collector_configupdate() {

-    if [ -z ${CONSUL_HOST} ] || [ -z ${CONFIG_BINDING_SERVICE} ] || [ -z ${HOSTNAME} ]; then

-        log "Using standard controller configuration (no dynamic configuration done)"

-    else

-        ${JAVA_HOME}/bin/java -cp "etc:lib/*" org.onap.dcae.controller.FetchDynamicConfig $*

-

-        if [ $? -ne 0 ]; then

-            logWarn "Failed to fetch dynamic configuration from consul into container /opt/app/KV-Configuration.json"

-        else

-            log "Dynamic config fetched and written successfully into container /opt/app/KV-Configuration.json"

-        fi

-

-        if [ -f /opt/app/KV-Configuration.json ]; then

-            ${JAVA_HOME}/bin/java -cp "etc:lib/*" org.onap.dcae.controller.LoadDynamicConfig $*

-            if [ $? -ne 0 ]; then

-                echo "ERROR: Failed to update dynamic configuration into Application"

-            else

-                echo "INFO: Dynamic config updated successfully into VESCollector configuration!"

-            fi

-            paramName="collector.keystore.alias"

-            localpropertyfile="etc/collector.properties"

-            tmpfile="etc/collector.properties.tmp"

-            keystore=`grep collector.keystore.file.location $localpropertyfile | tr -d '[:space:]' | cut -d"=" -f2`

-            keypwdfile=`grep collector.keystore.passwordfile $localpropertyfile | tr -d '[:space:]' | cut -d"=" -f2`

-            echo "/usr/bin/keytool -list -keystore $keystore < $keypwdfile | grep "PrivateKeyEntry" | cut -d"," -f1"

-            tmpalias=`/usr/bin/keytool -list -keystore $keystore < $keypwdfile | grep "PrivateKeyEntry" | cut -d"," -f1`

-            alias=`echo $tmpalias | cut -d":" -f2`

-            sed "s~$paramName=.*~$paramName=$alias~g" $localpropertyfile > $tmpfile

-            echo `cat $tmpfile > $localpropertyfile`

-            rm $tmpfile

-            log "Keystore alias updated"

-        else

-            logWarn "Configuration file /opt/app/KV-Configuration.json missing"

-        fi

-    fi

-}

-

-case $1 in

-  "start") collector_configupdate; start ;;

-  "stop")  stop ;;

-  *)       echo "Bad usage. Should be: /bin/bash <this> start/stop"

-esac

-

diff --git a/src/main/scripts/appController.sh b/src/main/scripts/appController.sh
new file mode 100644
index 00000000..d141addf
--- /dev/null
+++ b/src/main/scripts/appController.sh
@@ -0,0 +1,86 @@
+#!/bin/bash
+
+###
+# ============LICENSE_START=======================================================
+# PROJECT
+# ================================================================================
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# Copyright (C) 2018 Nokia Networks 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=========================================================
+###
+source bin/logger.sh
+
+updateKeystore() {
+    log "Updating keystore configuration"
+    aliasParameterName="collector.keystore.alias"
+    originalPropertyFile="etc/collector.properties"
+    temporaryPropertyFile="etc/collector.properties.tmp"
+    keystorePath=`grep collector.keystore.file.location ${originalPropertyFile} | tr -d '[:space:]' | cut -d"=" -f2`
+    keystorePasswordFile=`grep collector.keystore.passwordfile ${originalPropertyFile} | tr -d '[:space:]' | cut -d"=" -f2`
+    temporaryAlias=`/usr/bin/keytool -list -keystore $keystorePath < $keystorePasswordFile | grep "PrivateKeyEntry" | cut -d"," -f1`
+    newAlias=`echo $temporaryAlias | cut -d":" -f2`
+    sed "s~$aliasParameterName=.*~$aliasParameterName=$newAlias~g" ${originalPropertyFile} > ${temporaryPropertyFile}
+    echo `cat ${temporaryPropertyFile} > ${originalPropertyFile}`
+    rm ${temporaryPropertyFile}
+    log "Keystore configuration updated"
+}
+
+tryToPollConfiguration() {
+    log "Trying to poll configuration from CBS before application starts"
+    ${JAVA_HOME}/bin/java -cp "etc:lib/*" org.onap.dcae.controller.PreAppStartupConfigUpdater
+}
+
+start() {
+    log "Starting application"
+    appPids=`pidof java`
+
+    if [ ! -z ${appPids} ]; then
+        logWarn "Tried to start an application, but it is already running on PID(s): ${appPids}. Startup aborted."
+        exit 1
+    fi
+
+    ${JAVA_HOME}/bin/java -cp "etc:lib/*" \
+      -Xms256m -Xmx512m \
+      -XX:ErrorFile=logs/java_error%p.log \
+      -XX:+HeapDumpOnOutOfMemoryError \
+      -Dhttps.protocols=TLSv1.1,TLSv1.2 \
+      org.onap.dcae.VesApplication $* & &>> logs/collector.log
+}
+
+stop() {
+    log "Stopping application"
+    appPids=`pidof java`
+
+    if [ ! -z ${appPids} ]; then
+        echo "Killing java PID(s): ${appPids}"
+        kill -9 ${appPids}
+        sleep 5
+        if [ ! $(pidof java) ]; then
+            log "Application stopped"
+        else
+            logWarn "Application did not stop after 5 seconds"
+        fi
+    else
+        logWarn "Tried to stop an application, but it was not running";
+    fi
+}
+
+case $1 in
+  "start")    tryToPollConfiguration; updateKeystore; start ;;
+  "stop")     stop ;;
+  "restart")  stop; start ;;
+  *)          echo "Bad usage. Should be: /bin/bash <this> start/stop"
+esac
+
diff --git a/src/main/scripts/configurationPoller.sh b/src/main/scripts/configurationPoller.sh
new file mode 100644
index 00000000..59dbf840
--- /dev/null
+++ b/src/main/scripts/configurationPoller.sh
@@ -0,0 +1,46 @@
+#!/bin/bash
+###
+# ============LICENSE_START=======================================================
+# PROJECT
+# ================================================================================
+# Copyright (C) 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=========================================================
+###
+source bin/logger.sh
+
+# This scripts job is to continuously run in background and watch for changes in collector.properties
+# and in case it has changed, restart application.
+# collector.properties (and DmaapConfig.json) is being updated periodically by calling for configuration from CBS and it is
+# done inside the VESCollector application itself.
+# Configuration poller can be run regardless of deployment type.
+# It will always check for changes in collector.properties and in deployment scenario,
+# where dynamic configuration should not be used, necessary environment
+# variables that are needed (consul host, cbs name, app name) will be missing, and java app will
+# not update the configuration files so restart won't be triggered.
+
+# Start after a while, because once the application starts, it might happen that
+# it fetched new configuration. In that case, the application will already be started with newest config, there would
+# be no point in restarting it once again.
+sleep 2m
+
+while true
+do
+    sleep 1m
+    if [[ $(find etc/collector.properties -mmin -1 -print) ]]; then
+        log "Found change in collector.properties, updating keystore and restarting application"
+        bin/appController.sh restart
+    fi
+done
+
diff --git a/src/main/scripts/docker-entry.sh b/src/main/scripts/docker-entry.sh
index 0aad7584..c17dd958 100644
--- a/src/main/scripts/docker-entry.sh
+++ b/src/main/scripts/docker-entry.sh
@@ -1,66 +1,63 @@
-#!/bin/bash

-###

-# ============LICENSE_START=======================================================

-# PROJECT

-# ================================================================================

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

-# Copyright (C) 2018 Nokia Networks 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=========================================================

-###

-source bin/logger.sh

-

-# Redirect all stdout & stderr to a main log file, but also let it print into the console

-# At the time this script is invoked, these directories and files do not exist yet, so we need to create them

-mkdir -p logs

-touch logs/collector.log

-exec &> >(tee -a logs/collector.log)

-

-log "Enabling log rotation for collector.log"

-loggedCommand "cp etc/logrotate.conf /etc/logrotate.d"

-echo "* *	* * *	root    logrotate /etc/logrotate.conf" >> /etc/crontab

-log "Restarting cron"

-loggedCommand "service cron reload"

-loggedCommand "service cron start"

-

-log "Main application entry-point invoked"

-

-if [ ! -z ${COLLECTOR_IP} ]; then

-    log "Collector ip (${COLLECTOR_IP}) (env var 'COLLECTOR_IP') found, adding entry to /etc/hosts"

-    echo ${COLLECTOR_IP}  $(hostname).dcae.simpledemo.onap.org >> /etc/hosts

-fi

-

-if [ ! -z ${DMAAPHOST} ]; then

-    if [ -z "$(echo ${DMAAPHOST} | sed -e 's/[0-9\.]//g')" ]; then

-        log "DMaaP host (${DMAAPHOST}) (env var 'DMAAPHOST') found, adding entry to /etc/hosts"

-        echo "${DMAAPHOST}  onap-dmaap" >> /etc/hosts

-    else

-        log "DMaaP host (${DMAAPHOST}) (env var 'DMAAPHOST') found, adding entry to /etc/host.aliases"

-        echo "onap-dmaap ${DMAAPHOST}" >> /etc/host.aliases

-    fi

-else

-	logWarn "DMaaP host (env var 'DMAAPHOST') is missing. Events will not be published to DMaaP"

-fi

-

-log "Scheduling application to be started, looping indefinitely to hold the docker process"

-bin/VESrestfulCollector.sh stop

-bin/VESrestfulCollector.sh start &

-

-# Add below if config polling should be enabled. More specific to K8 deployment in ONAP

-if [ ! -z ${CBSPOLLTIMER} ]; then

-    log "Configuration poll time (${CBSPOLLTIMER}) (env var 'CBSPOLLTIMER') found, enabling configuration polling from CBS"

-    bin/VESConfigPoller.sh &

-fi

-

-while true; do sleep 1000; done

+#!/bin/bash
+###
+# ============LICENSE_START=======================================================
+# PROJECT
+# ================================================================================
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# Copyright (C) 2018 Nokia Networks 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=========================================================
+###
+source bin/logger.sh
+
+# Redirect all stdout & stderr to a main log file, but also let it print into the console
+# At the time this script is invoked, these directories and files do not exist yet, so we need to create them
+mkdir -p logs
+touch logs/collector.log
+exec &> >(tee -a logs/collector.log)
+
+log "Enabling log rotation for collector.log"
+loggedCommand "cp etc/logrotate.conf /etc/logrotate.d"
+echo "* *	* * *	root    logrotate /etc/logrotate.conf" >> /etc/crontab
+log "Restarting cron"
+loggedCommand "service cron reload"
+loggedCommand "service cron start"
+
+log "Main application entry-point invoked"
+
+if [ ! -z ${COLLECTOR_IP} ]; then
+    log "Collector ip (${COLLECTOR_IP}) (env var 'COLLECTOR_IP') found, adding entry to /etc/hosts"
+    echo ${COLLECTOR_IP}  $(hostname).dcae.simpledemo.onap.org >> /etc/hosts
+fi
+
+if [ ! -z ${DMAAPHOST} ]; then
+    if [ -z "$(echo ${DMAAPHOST} | sed -e 's/[0-9\.]//g')" ]; then
+        log "DMaaP host (${DMAAPHOST}) (env var 'DMAAPHOST') found, adding entry to /etc/hosts"
+        echo "${DMAAPHOST}  onap-dmaap" >> /etc/hosts
+    else
+        log "DMaaP host (${DMAAPHOST}) (env var 'DMAAPHOST') found, adding entry to /etc/host.aliases"
+        echo "onap-dmaap ${DMAAPHOST}" >> /etc/host.aliases
+    fi
+else
+	logWarn "DMaaP host (env var 'DMAAPHOST') is missing. Events will not be published to DMaaP"
+fi
+
+log "Scheduling application to be started, looping indefinitely to hold the docker process"
+bin/appController.sh stop
+bin/appController.sh start &
+
+log "Enabling configuration polling from CBS"
+bin/configurationPoller.sh &
+
+while true; do sleep 1000; done
diff --git a/src/test/java/org/onap/dcae/ApplicationSettingsTest.java b/src/test/java/org/onap/dcae/ApplicationSettingsTest.java
index b483bcb6..2ac42080 100644
--- a/src/test/java/org/onap/dcae/ApplicationSettingsTest.java
+++ b/src/test/java/org/onap/dcae/ApplicationSettingsTest.java
@@ -149,6 +149,26 @@ public class ApplicationSettingsTest {
     }
 
     @Test
+    public void shouldReturnConfigurationUpdateInterval() throws IOException {
+        // when
+        int updateFrequency = fromTemporaryConfiguration("collector.dynamic.config.update.frequency=10")
+                .configurationUpdateFrequency();
+
+        // then
+        assertEquals(10, updateFrequency);
+    }
+
+    @Test
+    public void shouldReturnDefaultConfigurationUpdateInterval() throws IOException {
+        // when
+        int updateFrequency = fromTemporaryConfiguration()
+                .configurationUpdateFrequency();
+
+        // then
+        assertEquals(5, updateFrequency);
+    }
+
+    @Test
     public void shouldReturnLocationOfThePasswordFile() throws IOException {
         // when
         String passwordFileLocation = fromTemporaryConfiguration("collector.keystore.passwordfile=/somewhere/password").keystorePasswordFileLocation();
@@ -207,7 +227,7 @@ public class ApplicationSettingsTest {
     @Test
     public void shouldReturnDMAAPConfigFileLocation() throws IOException {
         // when
-        String dmaapConfigFileLocation = fromTemporaryConfiguration("collector.dmaapfile=/somewhere/dmaapFile").cambriaConfigurationFileLocation();
+        String dmaapConfigFileLocation = fromTemporaryConfiguration("collector.dmaapfile=/somewhere/dmaapFile").dMaaPConfigurationFileLocation();
 
         // then
         assertEquals(sanitizePath("/somewhere/dmaapFile"), dmaapConfigFileLocation);
@@ -216,7 +236,7 @@ public class ApplicationSettingsTest {
     @Test
     public void shouldReturnDefaultDMAAPConfigFileLocation() throws IOException {
         // when
-        String dmaapConfigFileLocation = fromTemporaryConfiguration().cambriaConfigurationFileLocation();
+        String dmaapConfigFileLocation = fromTemporaryConfiguration().dMaaPConfigurationFileLocation();
 
         // then
         assertEquals(sanitizePath("etc/DmaapConfig.json"), dmaapConfigFileLocation);
@@ -390,7 +410,7 @@ public class ApplicationSettingsTest {
     public void shouldReturnCambriaConfigurationFileLocation() throws IOException {
         // when
         String cambriaConfigurationFileLocation = fromTemporaryConfiguration("collector.dmaapfile=/somewhere/dmaapConfig")
-                .cambriaConfigurationFileLocation();
+                .dMaaPConfigurationFileLocation();
 
         // then
         assertEquals(sanitizePath("/somewhere/dmaapConfig"), cambriaConfigurationFileLocation);
@@ -400,7 +420,7 @@ public class ApplicationSettingsTest {
     public void shouldReturnDefaultCambriaConfigurationFileLocation() throws IOException {
         // when
         String cambriaConfigurationFileLocation = fromTemporaryConfiguration()
-                .cambriaConfigurationFileLocation();
+                .dMaaPConfigurationFileLocation();
 
         // then
         assertEquals(sanitizePath("etc/DmaapConfig.json"), cambriaConfigurationFileLocation);
diff --git a/src/test/java/org/onap/dcae/TestingUtilities.java b/src/test/java/org/onap/dcae/TestingUtilities.java
new file mode 100644
index 00000000..0bbb6cc3
--- /dev/null
+++ b/src/test/java/org/onap/dcae/TestingUtilities.java
@@ -0,0 +1,94 @@
+/*-
+ * ============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;
+
+import static java.nio.file.Files.readAllBytes;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import io.vavr.control.Try;
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.assertj.core.api.AbstractThrowableAssert;
+import org.assertj.core.api.Java6Assertions;
+import org.json.JSONObject;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public final class TestingUtilities {
+
+    private TestingUtilities() {
+        // utility class, no objects allowed
+    }
+
+    public static void assertJSONObjectsEqual(JSONObject o1, JSONObject o2) {
+        assertThat(o1.toString()).isEqualTo(o2.toString());
+    }
+
+    public static JSONObject readJSONFromFile(Path path) {
+        return rethrow(() -> new JSONObject(readFile(path)));
+    }
+
+    public static String readFile(Path path) {
+        return rethrow(() -> new String(readAllBytes(path)));
+    }
+
+    public static Path createTemporaryFile(String content) {
+        return rethrow(() -> {
+            File temp = File.createTempFile("ves-collector-tests-created-this-file", ".tmp");
+            temp.deleteOnExit();
+            Path filePath = Paths.get(temp.toString());
+            Files.write(filePath, content.getBytes());
+            return filePath;
+        });
+    }
+
+    /**
+     * Exception in test case usually means there is something wrong, it should never be catched, but rather thrown to
+     * be handled by JUnit framework.
+     */
+    private static <T> T rethrow(CheckedSupplier<T> supplier) {
+        try {
+            return supplier.get();
+        } catch (Exception e) {
+            throw new RuntimeException();
+        }
+    }
+
+    @FunctionalInterface
+    interface CheckedSupplier<T> {
+
+        T get() throws Exception;
+    }
+
+
+    public static void assertFailureHasInfo(Try any, String... msgPart) {
+        Java6Assertions.assertThat(any.isFailure()).isTrue();
+        AbstractThrowableAssert<?, ? extends Throwable> o = Java6Assertions.assertThat(any.getCause())
+            .hasCauseInstanceOf(Exception.class);
+        for (String s : msgPart) {
+            o.hasStackTraceContaining(s);
+        }
+    }
+}
diff --git a/src/test/java/org/onap/dcae/WiremockBasedTest.java b/src/test/java/org/onap/dcae/WiremockBasedTest.java
new file mode 100644
index 00000000..ae851259
--- /dev/null
+++ b/src/test/java/org/onap/dcae/WiremockBasedTest.java
@@ -0,0 +1,68 @@
+/*-
+ * ============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;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static io.vavr.API.Map;
+
+import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import io.vavr.collection.Map;
+import org.junit.Rule;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public class WiremockBasedTest {
+
+    @Rule
+    public WireMockRule wireMockRule = new WireMockRule(
+        wireMockConfig().dynamicPort().dynamicHttpsPort().keystorePath(null));
+
+    protected void stubConsulToReturnLocalAddressOfCBS() {
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withBody(validLocalCBSConf())));
+    }
+
+    protected void stubCBSToReturnAppConfig(String sampleConfigForVES) {
+        stubFor(get(urlEqualTo("/service_component/VESCollector"))
+            .willReturn(aResponse().withBody(sampleConfigForVES)));
+    }
+
+    protected Map<String, String> wiremockBasedEnvProps() {
+        return Map(
+            "CONSUL_HOST", "http://localhost",
+            "CONSUL_PORT", "" + wireMockRule.port(),
+            "HOSTNAME", "VESCollector",
+            "CONFIG_BINDING_SERVICE", "CBSName"
+        );
+    }
+
+    protected String validLocalCBSConf() {
+        return ""
+            + "[{ "
+            + "\"ServiceAddress\": \"http://localhost\","
+            + "\"ServicePort\":" + wireMockRule.port()
+            + "}]";
+    }
+}
diff --git a/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java b/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java
new file mode 100644
index 00000000..336788a9
--- /dev/null
+++ b/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java
@@ -0,0 +1,151 @@
+/*-
+ * ============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 com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static org.assertj.core.api.Java6Assertions.assertThat;
+import static org.onap.dcae.TestingUtilities.assertFailureHasInfo;
+import static org.onap.dcae.controller.ConfigSource.getAppConfig;
+
+import io.vavr.control.Try;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.onap.dcae.WiremockBasedTest;
+
+public class ConfigCBSSourceTest extends WiremockBasedTest {
+
+    @Test
+    public void shouldReturnValidAppConfiguration() {
+        // given
+        String sampleConfigForVES = "{\"collector.port\": 8080}";
+
+        stubConsulToReturnLocalAddressOfCBS();
+        stubCBSToReturnAppConfig(sampleConfigForVES);
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertThat(actual.get().toString()).isEqualTo(new JSONObject(sampleConfigForVES).toString());
+    }
+
+    @Test
+    public void shouldReturnFailureOnFailureToCommunicateWithConsul() {
+        // given
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withStatus(400)));
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "HTTP", "Consul", "400",
+            "http://localhost:" + wireMockRule.port() + "/v1/catalog/service/CBSName");
+    }
+
+    @Test
+    public void shouldReturnFailureOnBadJsonFromConsul() {
+        // given
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withStatus(200).withBody("[{")));
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "JSON", "array", "[{");
+    }
+
+    @Test
+    public void shouldReturnFailureOnInvalidCatalogFormat() {
+        // given
+        String notAListCatalog = ""
+            + "{"
+            + "\"ServiceAddress\":\"http://localhost\","
+            + "\"ServicePort\":" + wireMockRule.port()
+            + "}";
+
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withStatus(200).withBody(notAListCatalog)));
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "JSON", "array", notAListCatalog);
+    }
+
+
+    @Test
+    public void shouldReturnFailureIfConfigIsMissingRequiredProperties() {
+        // given
+        String actualConf = "{\"ServicePort\":" + wireMockRule.port() + "}";
+        String asCatalog = "[" + actualConf + "]";
+
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withStatus(200).withBody(asCatalog)));
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "ServiceAddress", "ServicePort", "missing", actualConf);
+    }
+
+
+    @Test
+    public void shouldReturnFailureOnFailureToCommunicateWithCBS() {
+        // given
+        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
+            .willReturn(aResponse().withStatus(200).withBody(validLocalCBSConf())));
+        stubFor(get(urlEqualTo("/service_component/VESCollector"))
+            .willReturn(aResponse().withStatus(400)));
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "HTTP", "CBS", "400",
+            "http://localhost:" + wireMockRule.port() + "/service_component/VESCollector");
+    }
+
+    @Test
+    public void shouldReturnFailureIfAppIsInvalidJsonDocument() {
+        // given
+        String invalidAppConf = "[$";
+        stubConsulToReturnLocalAddressOfCBS();
+        stubCBSToReturnAppConfig(invalidAppConf);
+
+        // when
+        Try<JSONObject> actual = tryToGetConfig();
+
+        // then
+        assertFailureHasInfo(actual, "JSON", "document", invalidAppConf);
+    }
+
+    private Try<JSONObject> tryToGetConfig() {
+        return getAppConfig(new EnvProps("http://localhost", wireMockRule.port(), "CBSName", "VESCollector"));
+    }
+}
+
diff --git a/src/test/java/org/onap/dcae/controller/ConfigFilesFacadeTest.java b/src/test/java/org/onap/dcae/controller/ConfigFilesFacadeTest.java
new file mode 100644
index 00000000..474a77c5
--- /dev/null
+++ b/src/test/java/org/onap/dcae/controller/ConfigFilesFacadeTest.java
@@ -0,0 +1,139 @@
+/*-
+ * ============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.Map;
+import static io.vavr.API.Some;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcae.TestingUtilities.assertFailureHasInfo;
+import static org.onap.dcae.TestingUtilities.assertJSONObjectsEqual;
+import static org.onap.dcae.TestingUtilities.createTemporaryFile;
+import static org.onap.dcae.TestingUtilities.readFile;
+import static org.onap.dcae.TestingUtilities.readJSONFromFile;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.json.JSONObject;
+import org.junit.Test;
+
+public class ConfigFilesFacadeTest {
+
+    private static final Path NON_EXISTENT = Paths.get("/non-existent");
+    private static final ConfigFilesFacade TO_NON_EXISTENT_POINTING_FACADE = new ConfigFilesFacade(NON_EXISTENT,
+        NON_EXISTENT);
+
+    @Test
+    public void shouldReadPropertyFile() {
+        // given
+        Path temporaryFile = createTemporaryFile("some.property=10");
+
+        // when
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(temporaryFile, temporaryFile);
+
+        Try<Map<String, String>> propertiesConfigurations = configFilesFacade.readCollectorProperties();
+
+        // then
+        assertThat(propertiesConfigurations.isSuccess()).isTrue();
+        assertThat(propertiesConfigurations.get().containsKey("some.property")).isTrue();
+        assertThat(propertiesConfigurations.get().get("some.property")).isEqualTo(Some("10"));
+    }
+
+
+    @Test
+    public void shouldReadDMaaPFile() {
+        // given
+        Path temporaryFile = createTemporaryFile("{}");
+
+        // when
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(temporaryFile, temporaryFile);
+
+        Try<JSONObject> dMaaPConfiguration = configFilesFacade.readDMaaPConfiguration();
+
+        // then
+        assertThat(dMaaPConfiguration.isSuccess()).isTrue();
+        assertThat(dMaaPConfiguration.get().toString()).isEqualTo("{}");
+    }
+
+    @Test
+    public void shouldWriteDMaaPConf() {
+        // given
+        Path temporaryFile = createTemporaryFile("{}");
+        JSONObject desiredConf = new JSONObject("{\"key\": 1}");
+
+        // when
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(temporaryFile, temporaryFile);
+
+        Try<Void> propertiesConfigurations = configFilesFacade.writeDMaaPConfiguration(desiredConf);
+
+        // then
+        assertThat(propertiesConfigurations.isSuccess()).isTrue();
+        assertJSONObjectsEqual(readJSONFromFile(temporaryFile), desiredConf);
+    }
+
+
+    @Test
+    public void shouldWriteProperties() {
+        // given
+        Path temporaryFile = createTemporaryFile("{}");
+
+        // when
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(temporaryFile, temporaryFile);
+        Try<Void> propertiesConfigurations = configFilesFacade.writeProperties(Map("prop1", "hi"));
+
+        // then
+        assertThat(propertiesConfigurations.isSuccess()).isTrue();
+        assertThat(readFile(temporaryFile).trim()).isEqualTo("prop1 = hi");
+    }
+
+    @Test
+    public void shouldContainPropertiesPathInCaseOfFailures() {
+        Try<Map<String, String>> result = TO_NON_EXISTENT_POINTING_FACADE.readCollectorProperties();
+        assertThat(result.isFailure()).isTrue();
+        assertFailureHasInfo(result, NON_EXISTENT.toString());
+    }
+
+    @Test
+    public void shouldContainDMaaPPathPathInCaseOfFailures() {
+        Try<JSONObject> result = TO_NON_EXISTENT_POINTING_FACADE.readDMaaPConfiguration();
+        assertThat(result.isFailure()).isTrue();
+        assertFailureHasInfo(result, NON_EXISTENT.toString());
+    }
+
+    @Test
+    public void shouldContainPropertiesPathPathInCaseOfFailuresOnWrite() {
+        // given
+        Try<Void> result = TO_NON_EXISTENT_POINTING_FACADE.writeProperties(Map("any", "any"));
+        assertThat(result.isFailure()).isTrue();
+        assertFailureHasInfo(result, NON_EXISTENT.toString());
+    }
+
+    @Test
+    public void shouldContainDMaaPPathPathInCaseOfFailuresOnWrite() {
+        // given
+        Try<Void> result = TO_NON_EXISTENT_POINTING_FACADE.writeDMaaPConfiguration(new JSONObject());
+        assertThat(result.isFailure()).isTrue();
+        assertFailureHasInfo(result, NON_EXISTENT.toString());
+    }
+
+}
+
diff --git a/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java b/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java
new file mode 100644
index 00000000..09dca676
--- /dev/null
+++ b/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java
@@ -0,0 +1,92 @@
+/*-
+ * ============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.Map;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+import static org.onap.dcae.TestingUtilities.createTemporaryFile;
+import static org.onap.dcae.TestingUtilities.readFile;
+import static org.onap.dcae.TestingUtilities.readJSONFromFile;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.onap.dcae.WiremockBasedTest;
+import org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser;
+import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
+
+public class ConfigLoaderIntegrationE2ETest extends WiremockBasedTest {
+
+    @Test
+    public void testSuccessfulE2EFlow() {
+        // given
+        Path dMaaPConfigFile = createTemporaryFile("{}");
+        Path collectorPropertiesFile = createTemporaryFile("");
+        Path dMaaPConfigSource = Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json");
+        JSONObject dMaaPConf = readJSONFromFile(dMaaPConfigSource);
+        stubConsulToReturnLocalAddressOfCBS();
+        stubCBSToReturnAppConfig(f("{\"collector.port\": 8080, \"streams_publishes\": %s}}", dMaaPConf));
+
+        EventPublisher eventPublisherMock = mock(EventPublisher.class);
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(dMaaPConfigFile, collectorPropertiesFile);
+
+        // when
+        ConfigLoader configLoader = new ConfigLoader(eventPublisherMock::reconfigure, configFilesFacade, ConfigSource::getAppConfig, () -> wiremockBasedEnvProps());
+        configLoader.updateConfig();
+
+        // then
+        assertThat(readJSONFromFile(dMaaPConfigSource).toString()).isEqualTo(dMaaPConf.toString());
+        assertThat(readFile(collectorPropertiesFile).trim()).isEqualTo("collector.port = 8080");
+        verify(eventPublisherMock, times(1)).reconfigure(
+            DMaaPConfigurationParser.parseToDomainMapping(dMaaPConf).get()
+        );
+    }
+
+    @Test
+    public void shouldNotReconfigureNotOverwriteIfConfigurationHasNotChanged() {
+        // given
+        Path dMaaPConfigFile = createTemporaryFile("{}");
+        Path collectorPropertiesFile = createTemporaryFile("");
+        JSONObject dMaaPConf = readJSONFromFile(Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json"));
+        stubConsulToReturnLocalAddressOfCBS();
+        stubCBSToReturnAppConfig(f("{\"collector.port\": 8080, \"streams_publishes\": %s}}", dMaaPConf));
+        EventPublisher eventPublisherMock = mock(EventPublisher.class);
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(dMaaPConfigFile, collectorPropertiesFile);
+        configFilesFacade.writeProperties(Map("collector.port", "8080"));
+        configFilesFacade.writeDMaaPConfiguration(dMaaPConf);
+
+        // when
+        ConfigLoader configLoader = new ConfigLoader(eventPublisherMock::reconfigure, configFilesFacade, ConfigSource::getAppConfig, () -> wiremockBasedEnvProps());
+        configLoader.updateConfig();
+
+        // then
+        verifyZeroInteractions(eventPublisherMock);
+    }
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/ConfigParsingTest.java b/src/test/java/org/onap/dcae/controller/ConfigParsingTest.java
new file mode 100644
index 00000000..a00a3d3b
--- /dev/null
+++ b/src/test/java/org/onap/dcae/controller/ConfigParsingTest.java
@@ -0,0 +1,72 @@
+/*-
+ * ============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.Map;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcae.TestingUtilities.assertJSONObjectsEqual;
+import static org.onap.dcae.TestingUtilities.readJSONFromFile;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import java.nio.file.Paths;
+import org.json.JSONObject;
+import org.junit.Test;
+
+public class ConfigParsingTest {
+
+    @Test
+    public void shouldReturnDMaaPConfig() {
+        JSONObject dMaaPConf = readJSONFromFile(Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json"));
+        JSONObject root = new JSONObject();
+        root.put("key1", "someProperty");
+        root.put("key2", "someProperty");
+        root.put("streams_publishes", dMaaPConf);
+
+        Option<JSONObject> dMaaPConfig = ConfigParsing.getDMaaPConfig(root);
+
+        assertThat(dMaaPConfig.isEmpty()).isFalse();
+        assertJSONObjectsEqual(dMaaPConfig.get(), dMaaPConf);
+    }
+
+    @Test
+    public void shouldReturnEmptyIfDMaaPConfigIsInvalid() {
+        JSONObject root = new JSONObject();
+        root.put("streams_publishes", 1);
+
+        Option<JSONObject> dMaaPConfig = ConfigParsing.getDMaaPConfig(root);
+
+        assertThat(dMaaPConfig.isEmpty()).isTrue();
+    }
+
+    @Test
+    public void getProperties() {
+        JSONObject dMaaPConf = readJSONFromFile(Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json"));
+        JSONObject root = new JSONObject();
+        root.put("key1", "someProperty");
+        root.put("key2", "someProperty");
+        root.put("streams_publishes", dMaaPConf);
+
+        Map<String, String> properties = ConfigParsing.getProperties(root);
+        assertThat(properties).isEqualTo(Map("key1", "someProperty", "key2", "someProperty"));
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java b/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java
new file mode 100644
index 00000000..581f6eae
--- /dev/null
+++ b/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java
@@ -0,0 +1,68 @@
+/*-
+ * ============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.Map;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.onap.dcae.controller.EnvPropertiesReader.readEnvProps;
+
+import io.vavr.collection.Map;
+import org.junit.Test;
+
+
+public class EnvPropertiesReaderTest {
+
+    @Test
+    public void shouldReturnEmptyOnMissingConsulHost() {
+        Map<String, String> envs = Map(
+            "CONFIG_BINDING_SERVICE", "doesNotMatter",
+            "HOSTNAME", "doesNotMatter");
+        assertTrue(readEnvProps(envs).isEmpty());
+    }
+
+    @Test
+    public void shouldReturnEmptyOnMissingCBSName() {
+        Map<String, String> envs = Map(
+            "CONSUL_HOST", "doesNotMatter",
+            "HOSTNAME", "doesNotMatter");
+        assertTrue(readEnvProps(envs).isEmpty());
+    }
+
+    @Test
+    public void shouldReturnEmptyOnMissingVESAppName() {
+        Map<String, String> envs = Map(
+            "CONSUL_HOST", "doesNotMatter",
+            "CONFIG_BINDING_SERVICE", "doesNotMatter");
+        assertTrue(readEnvProps(envs).isEmpty());
+    }
+
+    @Test
+    public void shouldReturnSomeOfAllProperties() {
+        Map<String, String> envs = Map(
+            "CONSUL_HOST", "doesNotMatter",
+            "HOSTNAME", "doesNotMatter",
+            "CONFIG_BINDING_SERVICE", "doesNotMatter");
+        assertFalse(readEnvProps(envs).isEmpty());
+    }
+
+}
+
diff --git a/src/test/java/org/onap/dcae/vestest/TestFetchConfig.java b/src/test/java/org/onap/dcae/vestest/TestFetchConfig.java
deleted file mode 100644
index 0b6b5027..00000000
--- a/src/test/java/org/onap/dcae/vestest/TestFetchConfig.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*-

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

- * PROJECT

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

- * Copyright (C) 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.vestest;

-

-import static org.junit.Assert.assertTrue;

-import static org.onap.dcae.vestest.TestingUtilities.createTemporaryFile;

-

-import com.google.gson.JsonObject;

-import java.nio.file.Path;

-import org.junit.Before;

-import org.junit.Test;

-import org.onap.dcae.controller.FetchDynamicConfig;

-import org.onap.dcae.controller.LoadDynamicConfig;

-

-

-public class TestFetchConfig {

-

-    private Path temporaryFile;

-

-    @Before

-    public void setUp() {

-        temporaryFile = createTemporaryFile();

-    }

-

-    @Test

-    public void shouldWriteFileAndAttachDMaaPStreamsPropertiesFromConfiguration() {

-        // given

-        FetchDynamicConfig loadDynamicConfig = new FetchDynamicConfig();

-        FetchDynamicConfig.configFile = temporaryFile.toString();

-        String sampleConfiguration = LoadDynamicConfig.readFile("src/test/resources/controller-config_singleline_ip.json");

-

-        // when

-        loadDynamicConfig.writefile(sampleConfiguration);

-

-        // then

-        JsonObject actuallyWrittenJSONContent = TestingUtilities.readJSONFromFile(temporaryFile);

-        assertTrue(actuallyWrittenJSONContent.has("streams_publishes"));

-    }

-

-    @Test

-    public void shouldThrowNoErrorsWhileParsingConsulResponse() {

-        // given

-        FetchDynamicConfig.retString = "[{\"ID\":\"81bc2a17-8cfa-3f6f-30a9-a545a9b6ac2f\",\"Node\":\"zldcrdm5bdcc2dokr00\",\"Address\":\"135.25.108.161\",\"Datacenter\":\"zldcrdm5bdcc2\",\"TaggedAddresses\":{\"lan\":\"135.25.108.161\",\"wan\":\"135.25.108.161\"},\"NodeMeta\":{\"fqdn\":\"zldcrdm5bdcc2dokr00.2f3fb3.rdm5b.tci.att.com\"},\"ServiceID\":\"20299a144716:config_binding_service:10000\",\"ServiceName\":\"config_binding_service\",\"ServiceTags\":[],\"ServiceAddress\":\"135.25.108.161\",\"ServicePort\":10000,\"ServiceEnableTagOverride\":false,\"CreateIndex\":9153156,\"ModifyIndex\":9153156}]";

-

-        // then

-        FetchDynamicConfig.getCBS();

-    }

-

-

-    @Test

-    public void shouldReturnTrueOnConfigurationChange() {

-        // given

-        FetchDynamicConfig.configFile = "src/test/resources/controller-config_singleline_ip.json";

-        FetchDynamicConfig.retCBSString = "{\"header.authflag\": \"1\", \"collector.schema.file\": \"{\\\"v1\\\": \\\"./etc/CommonEventFormat_27.2.json\\\", \\\"v2\\\": \\\"./etc/CommonEventFormat_27.2.json\\\", \\\"v3\\\": \\\"./etc/CommonEventFormat_27.2.json\\\", \\\"v4\\\": \\\"./etc/CommonEventFormat_27.2.json\\\", \\\"v5\\\": \\\"./etc/CommonEventFormat_28.4.json\\\"}\", \"collector.keystore.passwordfile\": \"/opt/app/dcae-certificate/.password\", \"tomcat.maxthreads\": \"200\", \"collector.dmaap.streamid\": \"fault=ves-fault|syslog=ves-syslog|heartbeat=ves-heartbeat|measurementsForVfScaling=ves-measurement|mobileFlow=ves-mobileflow|other=ves-other|stateChange=ves-statechange|thresholdCrossingAlert=ves-thresholdCrossingAlert|voiceQuality=ves-voicequality|sipSignaling=ves-sipsignaling\", \"streams_subscribes\": {}, \"collector.inputQueue.maxPending\": \"8096\", \"collector.keystore.alias\": \"dynamically generated\", \"streams_publishes\": {\"ves-mobileflow\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590629043\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-MOBILEFLOW-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-measurement\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590433916\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-ENC-MEASUREMENT-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-voicequality\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590778397\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-VES-VOICEQUALITY-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-thresholdCrossingAlert\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590728150\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-TCA-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-fault\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590384670\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-FAULT-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-heartbeat\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590530041\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-HEARTBEAT-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-sipsignaling\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590828736\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-VES-SIPSIGNALING-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-syslog\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590482019\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-SYSLOG-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-other\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590581045\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-OTHER-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}, \"ves-statechange\": {\"type\": \"message_router\", \"dmaap_info\": {\"client_id\": \"1517590677649\", \"client_role\": \"com.att.secCollector.member\", \"location\": \"rdm5bdcc2\", \"topic_url\": \"https://DMAAPHOST:3905/events/com.att.dcae.dmaap.FTL.24256-SEC-STATECHANGE-OUTPUT-v1\"}, \"aaf_username\": \"userid@namespace\", \"aaf_password\": \"authpwd\"}}, \"collector.schema.checkflag\": \"1\", \"services_calls\": {}, \"event.transform.flag\": \"1\", \"collector.keystore.file.location\": \"/opt/app/dcae-certificate/keystore.jks\", \"header.authlist\": \"sample1,c2FtcGxlMQ==|userid1,base64encodepwd1|userid2,base64encodepwd2\", \"collector.service.secure.port\": \"8443\", \"collector.service.port\": \"-1\"}";

-

-        // when

-        boolean didConfigsChange = FetchDynamicConfig.verifyConfigChange();

-

-        // then

-        assertTrue(didConfigsChange);

-    }

-

-}

-

diff --git a/src/test/java/org/onap/dcae/vestest/TestLoadDynamicConfig.java b/src/test/java/org/onap/dcae/vestest/TestLoadDynamicConfig.java
deleted file mode 100644
index 03a074d7..00000000
--- a/src/test/java/org/onap/dcae/vestest/TestLoadDynamicConfig.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 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.vestest;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.onap.dcae.vestest.TestingUtilities.createTemporaryFile;
-
-import com.github.fge.jackson.JsonLoader;
-import com.google.gson.JsonObject;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import org.json.JSONObject;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.dcae.controller.LoadDynamicConfig;
-
-public class TestLoadDynamicConfig {
-
-    private Path temporaryFile;
-
-    @Before
-    public void setUp() {
-        temporaryFile = createTemporaryFile();
-    }
-
-    @Test
-    public void shouldReadFileContent() throws IOException {
-        // given
-        String expectedJSON = "{ \"field\" : 1 }";
-        Files.write(temporaryFile, expectedJSON.getBytes());
-
-        // when
-        String readFileContent = LoadDynamicConfig.readFile(temporaryFile.toString());
-
-        // then
-        assertEquals(JsonLoader.fromString(expectedJSON), JsonLoader.fromString(readFileContent));
-    }
-
-    @Test
-    public void shouldWriteFileAndAttachDMaaPRelatedPropertiesFromConfiguration() {
-        // given
-        LoadDynamicConfig loadDynamicConfig = new LoadDynamicConfig();
-        loadDynamicConfig.propFile = "src/test/resources/test_collector_ip_op.properties";
-        loadDynamicConfig.configFile = "src/test/resources/controller-config_dmaap_ip.json";
-        loadDynamicConfig.dMaaPOutputFile = temporaryFile.toString();
-        String sampleConfiguration = LoadDynamicConfig.readFile(loadDynamicConfig.configFile);
-
-        // when
-        loadDynamicConfig.writeconfig(new JSONObject(sampleConfiguration));
-
-        // then
-        JsonObject actuallyWrittenJSONContent = TestingUtilities.readJSONFromFile(temporaryFile);
-        assertTrue(actuallyWrittenJSONContent.has("ves-fault-secondary"));
-    }
-
-}
-