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
  26 import java.io.BufferedInputStream;
  27 import java.io.FileInputStream;
  28 import java.io.IOException;
  29 import java.io.InputStream;
  30 import java.io.InputStreamReader;
  31 import java.time.Duration;
  32 import java.util.Map;
  33 import java.util.Properties;
  34 import java.util.ServiceLoader;
  35
  36 import javax.validation.constraints.NotEmpty;
  37 import javax.validation.constraints.NotNull;
  38
  39 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  40 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
  41 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
  42 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.ImmutableEnvProperties;
  43 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46 import org.springframework.beans.factory.annotation.Autowired;
  47 import org.springframework.beans.factory.annotation.Value;
  48 import org.springframework.boot.context.properties.ConfigurationProperties;
  49 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  50 import org.springframework.context.annotation.ComponentScan;
  51 import org.springframework.stereotype.Component;
  52
  53 import reactor.core.Disposable;
  54 import reactor.core.publisher.Flux;
  55 import reactor.core.publisher.Mono;
  56
  57 /**
  58  * Holds all configuration for the DFC.
  59  *
  60  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  61  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  62  */
  63
  64 @Component
  65 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  66 @EnableConfigurationProperties
  67 @ConfigurationProperties("app")
  68 public class AppConfig {
  69     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
  70
  71     private ConsumerConfiguration dmaapConsumerConfiguration;
  72     private Map<String, PublisherConfiguration> publishingConfiguration;
  73     private FtpesConfig ftpesConfiguration;
  74     private CloudConfigurationProvider cloudConfigurationProvider;
  75     @Value("#{systemEnvironment}")
  76     Properties systemEnvironment;
  77     private Disposable refreshConfigTask = null;
  78
  79     @NotEmpty
  80     private String filepath;
  81
  82     @Autowired
  83     public synchronized void setCloudConfigurationProvider(
  84         CloudConfigurationProvider reactiveCloudConfigurationProvider) {
  85         this.cloudConfigurationProvider = reactiveCloudConfigurationProvider;
  86     }
  87
  88     public synchronized void setFilepath(String filepath) {
  89         this.filepath = filepath;
  90     }
  91
  92     /**
  93      * Reads the cloud configuration.
  94      */
  95     public void initialize() {
  96         stop();
  97         Map<String, String> context = MappedDiagnosticContext.initializeTraceContext();
  98         loadConfigurationFromFile();
  99
 100         refreshConfigTask = Flux.interval(Duration.ZERO, Duration.ofMinutes(5))
 101             .flatMap(count -> createRefreshConfigurationTask(count, context))
 102             .subscribe(e -> logger.info("Refreshed configuration data"),
 103                 throwable -> logger.error("Configuration refresh terminated due to exception", throwable),
 104                 () -> logger.error("Configuration refresh terminated"));
 105     }
 106
 107     public void stop() {
 108         if (refreshConfigTask != null) {
 109             refreshConfigTask.dispose();
 110             refreshConfigTask = null;
 111         }
 112     }
 113
 114     public synchronized ConsumerConfiguration getDmaapConsumerConfiguration() {
 115         return dmaapConsumerConfiguration;
 116     }
 117
 118     public synchronized boolean isFeedConfigured(String changeIdentifier) {
 119         return publishingConfiguration.containsKey(changeIdentifier);
 120     }
 121
 122     public synchronized PublisherConfiguration getPublisherConfiguration(String changeIdentifier)
 123         throws DatafileTaskException {
 124
 125         if (publishingConfiguration == null) {
 126             throw new DatafileTaskException("No PublishingConfiguration loaded, changeIdentifier: " + changeIdentifier);
 127         }
 128         PublisherConfiguration cfg = publishingConfiguration.get(changeIdentifier);
 129         if (cfg == null) {
 130             throw new DatafileTaskException(
 131                 "Cannot find getPublishingConfiguration for changeIdentifier: " + changeIdentifier);
 132         }
 133         return cfg;
 134     }
 135
 136     public synchronized FtpesConfig getFtpesConfiguration() {
 137         return ftpesConfiguration;
 138     }
 139
 140     Flux<AppConfig> createRefreshConfigurationTask(Long counter, Map<String, String> context) {
 141         return Flux.just(counter) //
 142             .doOnNext(cnt -> logger.debug("Refresh config {}", cnt)) //
 143             .flatMap(cnt -> readEnvironmentVariables(systemEnvironment, context)) //
 144             .flatMap(this::fetchConfiguration);
 145     }
 146
 147     Mono<EnvProperties> readEnvironmentVariables(Properties systemEnvironment, Map<String, String> context) {
 148         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context)
 149             .onErrorResume(AppConfig::onErrorResume);
 150     }
 151
 152     private static <R> Mono<R> onErrorResume(Throwable trowable) {
 153         logger.error("Could not refresh application configuration {}", trowable.toString());
 154         return Mono.empty();
 155     }
 156
 157     private Mono<AppConfig> fetchConfiguration(EnvProperties env) {
 158         Mono<JsonObject> serviceCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(env) //
 159             .onErrorResume(AppConfig::onErrorResume);
 160
 161         // Note, have to use this callForServiceConfigurationReactive with EnvProperties, since the
 162         // other ones does not work
 163         EnvProperties dmaapEnv = ImmutableEnvProperties.builder() //
 164             .consulHost(env.consulHost()) //
 165             .consulPort(env.consulPort()) //
 166             .cbsName(env.cbsName()) //
 167             .appName(env.appName() + ":dmaap") //
 168             .build(); //
 169         Mono<JsonObject> dmaapCfg = cloudConfigurationProvider.callForServiceConfigurationReactive(dmaapEnv)
 170             .onErrorResume(t -> Mono.just(new JsonObject()));
 171
 172         return serviceCfg.zipWith(dmaapCfg, this::parseCloudConfig) //
 173             .onErrorResume(AppConfig::onErrorResume);
 174     }
 175
 176     /**
 177      * Parse configuration.
 178      *
 179      * @param serviceConfigRootObject the DFC service's configuration
 180      * @param dmaapConfigRootObject if there is no dmaapConfigRootObject, the dmaap feeds are taken from the
 181      *        serviceConfigRootObject
 182      * @return this which is updated if successful
 183      */
 184     private AppConfig parseCloudConfig(JsonObject serviceConfigRootObject, JsonObject dmaapConfigRootObject) {
 185         try {
 186             CloudConfigParser parser = new CloudConfigParser(serviceConfigRootObject, dmaapConfigRootObject);
 187             setConfiguration(parser.getDmaapConsumerConfig(), parser.getDmaapPublisherConfig(),
 188                 parser.getFtpesConfig());
 189         } catch (DatafileTaskException e) {
 190             logger.error("Could not parse configuration {}", e.toString(), e);
 191         }
 192         return this;
 193     }
 194
 195     /**
 196      * Reads the configuration from file.
 197      */
 198     void loadConfigurationFromFile() {
 199         GsonBuilder gsonBuilder = new GsonBuilder();
 200         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 201
 202         try (InputStream inputStream = createInputStream(filepath)) {
 203             JsonParser parser = new JsonParser();
 204             JsonObject rootObject = getJsonElement(parser, inputStream).getAsJsonObject();
 205             if (rootObject == null) {
 206                 throw new JsonSyntaxException("Root is not a json object");
 207             }
 208             parseCloudConfig(rootObject, rootObject);
 209         } catch (JsonSyntaxException | IOException e) {
 210             logger.warn("Local configuration file not loaded: {}", filepath, e);
 211         }
 212     }
 213
 214     private synchronized void setConfiguration(ConsumerConfiguration consumerConfiguration,
 215         Map<String, PublisherConfiguration> publisherConfiguration, FtpesConfig ftpesConfig) {
 216         if (consumerConfiguration == null || publisherConfiguration == null || ftpesConfig == null) {
 217             logger.error(
 218                 "Problem with configuration consumerConfiguration: {}, publisherConfiguration: {}, ftpesConfig: {}",
 219                 consumerConfiguration, publisherConfiguration, ftpesConfig);
 220         } else {
 221             this.dmaapConsumerConfiguration = consumerConfiguration;
 222             this.publishingConfiguration = publisherConfiguration;
 223             this.ftpesConfiguration = ftpesConfig;
 224         }
 225     }
 226
 227     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
 228         return parser.parse(new InputStreamReader(inputStream));
 229     }
 230
 231     InputStream createInputStream(@NotNull String filepath) throws IOException {
 232         return new BufferedInputStream(new FileInputStream(filepath));
 233     }
 234
 235 }