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     private Map<String, PublisherConfiguration> publishingConfiguration;
  69     private FtpesConfig ftpesConfiguration;
  70     private CloudConfigurationProvider cloudConfigurationProvider;
  71     @Value("#{systemEnvironment}")
  72     Properties systemEnvironment;
  73     private 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     public void stop() {
 104         if (refreshConfigTask != null) {
 105             refreshConfigTask.dispose();
 106             refreshConfigTask = null;
 107         }
 108     }
 109
 110     public synchronized ConsumerConfiguration getDmaapConsumerConfiguration() {
 111         return dmaapConsumerConfiguration;
 112     }
 113
 114     public synchronized boolean isFeedConfigured(String changeIdentifier) {
 115         return publishingConfiguration.containsKey(changeIdentifier);
 116     }
 117
 118     public synchronized PublisherConfiguration getPublisherConfiguration(String changeIdentifier)
 119             throws DatafileTaskException {
 120
 121         if (publishingConfiguration == null) {
 122             throw new DatafileTaskException("No PublishingConfiguration loaded, changeIdentifier: " + changeIdentifier);
 123         }
 124         PublisherConfiguration cfg = publishingConfiguration.get(changeIdentifier);
 125         if (cfg == null) {
 126             throw new DatafileTaskException(
 127                     "Cannot find getPublishingConfiguration for changeIdentifier: " + changeIdentifier);
 128         }
 129         return cfg;
 130     }
 131
 132     public synchronized FtpesConfig getFtpesConfiguration() {
 133         return ftpesConfiguration;
 134     }
 135
 136     Flux<AppConfig> createRefreshConfigurationTask(Long counter, Map<String, String> context) {
 137         return Flux.just(counter) //
 138                 .doOnNext(cnt -> logger.debug("Refresh config {}", cnt)) //
 139                 .flatMap(cnt -> readEnvironmentVariables(systemEnvironment, context)) //
 140                 .flatMap(this::fetchConfiguration);
 141     }
 142
 143     Mono<EnvProperties> readEnvironmentVariables(Properties systemEnvironment, Map<String, String> context) {
 144         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context)
 145                 .onErrorResume(AppConfig::onErrorResume);
 146     }
 147
 148     private static <R> Mono<R> onErrorResume(Throwable trowable) {
 149         logger.error("Could not refresh application configuration {}", trowable.toString());
 150         return Mono.empty();
 151     }
 152
 153     private Mono<AppConfig> fetchConfiguration(EnvProperties env) {
 154         Mono<JsonObject> serviceCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(env) //
 155                 .onErrorResume(AppConfig::onErrorResume);
 156
 157         // Note, have to use this callForServiceConfigurationReactive with EnvProperties, since the
 158         // other ones does not work
 159         EnvProperties dmaapEnv = ImmutableEnvProperties.builder() //
 160                 .consulHost(env.consulHost()) //
 161                 .consulPort(env.consulPort()) //
 162                 .cbsName(env.cbsName()) //
 163                 .appName(env.appName() + ":dmaap") //
 164                 .build(); //
 165         Mono<JsonObject> dmaapCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(dmaapEnv)
 166                 .onErrorResume(t -> Mono.just(new JsonObject()));
 167
 168         return serviceCfg.zipWith(dmaapCfg, this::parseCloudConfig) //
 169                 .onErrorResume(AppConfig::onErrorResume);
 170     }
 171
 172     /**
 173      * Parse configuration.
 174      *
 175      * @param serviceConfigRootObject the DFC service's configuration
 176      * @param dmaapConfigRootObject if there is no dmaapConfigRootObject, the dmaap feeds are taken from the
 177      *        serviceConfigRootObject
 178      * @return this which is updated if successful
 179      */
 180     private AppConfig parseCloudConfig(JsonObject serviceConfigRootObject, JsonObject dmaapConfigRootObject) {
 181         try {
 182             CloudConfigParser parser = new CloudConfigParser(serviceConfigRootObject, dmaapConfigRootObject);
 183             setConfiguration(parser.getDmaapConsumerConfig(), parser.getDmaapPublisherConfig(),
 184                     parser.getFtpesConfig());
 185         } catch (DatafileTaskException e) {
 186             logger.error("Could not parse configuration {}", e.toString(), e);
 187         }
 188         return this;
 189     }
 190
 191     /**
 192      * Reads the configuration from file.
 193      */
 194     void loadConfigurationFromFile() {
 195         GsonBuilder gsonBuilder = new GsonBuilder();
 196         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 197
 198         try (InputStream inputStream = createInputStream(filepath)) {
 199             JsonParser parser = new JsonParser();
 200             JsonObject rootObject = getJsonElement(parser, inputStream).getAsJsonObject();
 201             if (rootObject == null) {
 202                 throw new JsonSyntaxException("Root is not a json object");
 203             }
 204             parseCloudConfig(rootObject, rootObject);
 205         } catch (JsonSyntaxException | IOException e) {
 206             logger.warn("Local configuration file not loaded: {}", filepath, e);
 207         }
 208     }
 209
 210     private synchronized void setConfiguration(ConsumerConfiguration consumerConfiguration,
 211             Map<String, PublisherConfiguration> publisherConfiguration, FtpesConfig ftpesConfig) {
 212         if (consumerConfiguration == null || publisherConfiguration == null || ftpesConfig == null) {
 213             logger.error(
 214                     "Problem with configuration consumerConfiguration: {}, publisherConfiguration: {}, ftpesConfig: {}",
 215                     consumerConfiguration, publisherConfiguration, ftpesConfig);
 216         } else {
 217             this.dmaapConsumerConfiguration = consumerConfiguration;
 218             this.publishingConfiguration = publisherConfiguration;
 219             this.ftpesConfiguration = ftpesConfig;
 220         }
 221     }
 222
 223     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
 224         return parser.parse(new InputStreamReader(inputStream));
 225     }
 226
 227     InputStream createInputStream(@NotNull String filepath) throws IOException {
 228         return new BufferedInputStream(new FileInputStream(filepath));
 229     }
 230
 231 }