gerrit.onap Code Review - dcaegen2/collectors/datafile.git/blob

   1 /*-
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
   4  * ===============================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   6  * in compliance with the License. You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software distributed under the License
  11  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  12  * or implied. See the License for the specific language governing permissions and limitations under
  13  * the License.
  14  * ============LICENSE_END========================================================================
  15  */
  16
  17 package org.onap.dcaegen2.collectors.datafile.configuration;
  18
  19 import com.google.gson.GsonBuilder;
  20 import com.google.gson.JsonElement;
  21 import com.google.gson.JsonObject;
  22 import com.google.gson.JsonParser;
  23 import com.google.gson.JsonSyntaxException;
  24 import com.google.gson.TypeAdapterFactory;
  25 import java.io.BufferedInputStream;
  26 import java.io.FileInputStream;
  27 import java.io.IOException;
  28 import java.io.InputStream;
  29 import java.io.InputStreamReader;
  30 import java.time.Duration;
  31 import java.util.Map;
  32 import java.util.Properties;
  33 import java.util.ServiceLoader;
  34 import javax.validation.constraints.NotEmpty;
  35 import javax.validation.constraints.NotNull;
  36 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  37 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
  38 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
  39 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.ImmutableEnvProperties;
  40 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43 import org.springframework.beans.factory.annotation.Autowired;
  44 import org.springframework.beans.factory.annotation.Value;
  45 import org.springframework.boot.context.properties.ConfigurationProperties;
  46 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  47 import org.springframework.context.annotation.ComponentScan;
  48 import org.springframework.stereotype.Component;
  49 import reactor.core.Disposable;
  50 import reactor.core.publisher.Flux;
  51 import reactor.core.publisher.Mono;
  52
  53 /**
  54  * Holds all configuration for the DFC.
  55  *
  56  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  57  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  58  */
  59
  60 @Component
  61 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  62 @EnableConfigurationProperties
  63 @ConfigurationProperties("app")
  64 public class AppConfig {
  65     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
  66
  67     private ConsumerConfiguration dmaapConsumerConfiguration;
  68     Map<String, PublisherConfiguration> publishingConfigurations;
  69     private FtpesConfig ftpesConfiguration;
  70     private CloudConfigurationProvider cloudConfigurationProvider;
  71     @Value("#{systemEnvironment}")
  72     Properties systemEnvironment;
  73     Disposable refreshConfigTask = null;
  74
  75     @NotEmpty
  76     private String filepath;
  77
  78     @Autowired
  79     public synchronized void setCloudConfigurationProvider(
  80         CloudConfigurationProvider reactiveCloudConfigurationProvider) {
  81         this.cloudConfigurationProvider = reactiveCloudConfigurationProvider;
  82     }
  83
  84     public synchronized void setFilepath(String filepath) {
  85         this.filepath = filepath;
  86     }
  87
  88     /**
  89      * Reads the cloud configuration.
  90      */
  91     public void initialize() {
  92         stop();
  93         Map<String, String> context = MappedDiagnosticContext.initializeTraceContext();
  94         loadConfigurationFromFile();
  95
  96         refreshConfigTask = Flux.interval(Duration.ZERO, Duration.ofMinutes(5))
  97             .flatMap(count -> createRefreshConfigurationTask(count, context))
  98             .subscribe(e -> logger.info("Refreshed configuration data"),
  99                 throwable -> logger.error("Configuration refresh terminated due to exception", throwable),
 100                 () -> logger.error("Configuration refresh terminated"));
 101     }
 102
 103     /**
 104      * Stops the refreshing of the configuration.
 105      */
 106     public void stop() {
 107         if (refreshConfigTask != null) {
 108             refreshConfigTask.dispose();
 109             refreshConfigTask = null;
 110         }
 111     }
 112
 113     public synchronized ConsumerConfiguration getDmaapConsumerConfiguration() {
 114         return dmaapConsumerConfiguration;
 115     }
 116
 117     /**
 118      * Checks if there is a configuration for the given feed.
 119      *
 120      * @param changeIdentifier the change identifier the feed is configured to belong to.
 121      *
 122      * @return true if a feed is configured for the given change identifier, false if not.
 123      */
 124     public synchronized boolean isFeedConfigured(String changeIdentifier) {
 125         return publishingConfigurations.containsKey(changeIdentifier);
 126     }
 127
 128     /**
 129      * Gets the feed configuration for the given change identifier.
 130      *
 131      * @param changeIdentifier the change identifier the feed is configured to belong to.
 132      * @return the <code>PublisherConfiguration</code> for the feed belonging to the given change identifier.
 133      *
 134      * @throws DatafileTaskException if no configuration has been loaded or the configuration is missing for the given
 135      *         change identifier.
 136      */
 137     public synchronized PublisherConfiguration getPublisherConfiguration(String changeIdentifier)
 138         throws DatafileTaskException {
 139
 140         if (publishingConfigurations == null) {
 141             throw new DatafileTaskException("No PublishingConfiguration loaded, changeIdentifier: " + changeIdentifier);
 142         }
 143         PublisherConfiguration cfg = publishingConfigurations.get(changeIdentifier);
 144         if (cfg == null) {
 145             throw new DatafileTaskException(
 146                 "Cannot find getPublishingConfiguration for changeIdentifier: " + changeIdentifier);
 147         }
 148         return cfg;
 149     }
 150
 151     public synchronized FtpesConfig getFtpesConfiguration() {
 152         return ftpesConfiguration;
 153     }
 154
 155     Flux<AppConfig> createRefreshConfigurationTask(Long counter, Map<String, String> context) {
 156         return Flux.just(counter) //
 157             .doOnNext(cnt -> logger.debug("Refresh config {}", cnt)) //
 158             .flatMap(cnt -> readEnvironmentVariables(systemEnvironment, context)) //
 159             .flatMap(this::fetchConfiguration);
 160     }
 161
 162     Mono<EnvProperties> readEnvironmentVariables(Properties systemEnvironment, Map<String, String> context) {
 163         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context)
 164             .onErrorResume(AppConfig::onErrorResume);
 165     }
 166
 167     private static <R> Mono<R> onErrorResume(Throwable trowable) {
 168         logger.error("Could not refresh application configuration {}", trowable.toString());
 169         return Mono.empty();
 170     }
 171
 172     private Mono<AppConfig> fetchConfiguration(EnvProperties env) {
 173         Mono<JsonObject> serviceCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(env) //
 174             .onErrorResume(AppConfig::onErrorResume);
 175
 176         // Note, have to use this callForServiceConfigurationReactive with EnvProperties, since the
 177         // other ones does not work
 178         EnvProperties dmaapEnv = ImmutableEnvProperties.builder() //
 179             .consulHost(env.consulHost()) //
 180             .consulPort(env.consulPort()) //
 181             .cbsName(env.cbsName()) //
 182             .appName(env.appName() + ":dmaap") //
 183             .build(); //
 184         Mono<JsonObject> dmaapCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(dmaapEnv)
 185             .onErrorResume(t -> Mono.just(new JsonObject()));
 186
 187         return serviceCfg.zipWith(dmaapCfg, this::parseCloudConfig) //
 188             .onErrorResume(AppConfig::onErrorResume);
 189     }
 190
 191     /**
 192      * Parse configuration.
 193      *
 194      * @param serviceConfigRootObject the DFC service's configuration
 195      * @param dmaapConfigRootObject if there is no dmaapConfigRootObject, the dmaap feeds are taken from the
 196      *        serviceConfigRootObject
 197      * @return this which is updated if successful
 198      */
 199     private AppConfig parseCloudConfig(JsonObject serviceConfigRootObject, JsonObject dmaapConfigRootObject) {
 200         try {
 201             CloudConfigParser parser = new CloudConfigParser(serviceConfigRootObject, dmaapConfigRootObject);
 202             setConfiguration(parser.getDmaapConsumerConfig(), parser.getDmaapPublisherConfigurations(),
 203                 parser.getFtpesConfig());
 204         } catch (DatafileTaskException e) {
 205             logger.error("Could not parse configuration {}", e.toString(), e);
 206         }
 207         return this;
 208     }
 209
 210     /**
 211      * Reads the configuration from file.
 212      */
 213     void loadConfigurationFromFile() {
 214         GsonBuilder gsonBuilder = new GsonBuilder();
 215         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 216
 217         try (InputStream inputStream = createInputStream(filepath)) {
 218             JsonParser parser = new JsonParser();
 219             JsonObject rootObject = getJsonElement(parser, inputStream).getAsJsonObject();
 220             if (rootObject == null) {
 221                 throw new JsonSyntaxException("Root is not a json object");
 222             }
 223             parseCloudConfig(rootObject, rootObject);
 224         } catch (JsonSyntaxException | IOException e) {
 225             logger.warn("Local configuration file not loaded: {}", filepath, e);
 226         }
 227     }
 228
 229     private synchronized void setConfiguration(ConsumerConfiguration consumerConfiguration,
 230         Map<String, PublisherConfiguration> publisherConfigurations, FtpesConfig ftpesConfig) {
 231         if (consumerConfiguration == null || publisherConfigurations == null || ftpesConfig == null) {
 232             logger.error("Problem with consumerConfiguration: {}, publisherConfigurations: {}, ftpesConfig: {}",
 233                 consumerConfiguration, publisherConfigurations, ftpesConfig);
 234         } else {
 235             this.dmaapConsumerConfiguration = consumerConfiguration;
 236             this.publishingConfigurations = publisherConfigurations;
 237             this.ftpesConfiguration = ftpesConfig;
 238         }
 239     }
 240
 241     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
 242         return parser.parse(new InputStreamReader(inputStream));
 243     }
 244
 245     InputStream createInputStream(@NotNull String filepath) throws IOException {
 246         return new BufferedInputStream(new FileInputStream(filepath));
 247     }
 248
 249 }