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.api.CbsClient;
  42 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
  43 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
  44 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
  45 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties;
  46 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49 import org.springframework.beans.factory.annotation.Value;
  50 import org.springframework.boot.context.properties.ConfigurationProperties;
  51 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  52 import org.springframework.context.annotation.ComponentScan;
  53 import org.springframework.stereotype.Component;
  54
  55 import reactor.core.Disposable;
  56 import reactor.core.publisher.Flux;
  57 import reactor.core.publisher.Mono;
  58
  59 /**
  60  * Holds all configuration for the DFC.
  61  *
  62  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  63  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  64  */
  65
  66 @Component
  67 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  68 @EnableConfigurationProperties
  69 @ConfigurationProperties("app")
  70 public class AppConfig {
  71
  72     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
  73
  74     private ConsumerConfiguration dmaapConsumerConfiguration;
  75     private Map<String, PublisherConfiguration> publishingConfigurations;
  76     private FtpesConfig ftpesConfiguration;
  77     private SftpConfig sftpConfiguration;
  78     @Value("#{systemEnvironment}")
  79     Properties systemEnvironment;
  80     private Disposable refreshConfigTask = null;
  81
  82     @NotEmpty
  83     private String filepath;
  84
  85     public synchronized void setFilepath(String filepath) {
  86         this.filepath = filepath;
  87     }
  88
  89     /**
  90      * Reads the cloud configuration.
  91      */
  92     public void initialize() {
  93         stop();
  94         Map<String, String> context = MappedDiagnosticContext.initializeTraceContext();
  95
  96         loadConfigurationFromFile();
  97
  98         refreshConfigTask = createRefreshTask(context) //
  99             .subscribe(e -> logger.info("Refreshed configuration data"),
 100                 throwable -> logger.error("Configuration refresh terminated due to exception", throwable),
 101                 () -> logger.error("Configuration refresh terminated"));
 102     }
 103
 104     Flux<AppConfig> createRefreshTask(Map<String, String> context) {
 105         return getEnvironment(systemEnvironment, context) //
 106             .flatMap(this::createCbsClient) //
 107             .flatMapMany(this::periodicConfigurationUpdates) //
 108             .map(this::parseCloudConfig) //
 109             .onErrorResume(this::onErrorResume);
 110     }
 111
 112     private Flux<JsonObject> periodicConfigurationUpdates(CbsClient cbsClient) {
 113         final Duration initialDelay = Duration.ZERO;
 114         final Duration refreshPeriod = Duration.ofMinutes(1);
 115         final CbsRequest getConfigRequest = CbsRequests.getAll(RequestDiagnosticContext.create());
 116         return cbsClient.updates(getConfigRequest, initialDelay, refreshPeriod);
 117     }
 118
 119     /**
 120      * Stops the refreshing of the configuration.
 121      */
 122     public void stop() {
 123         if (refreshConfigTask != null) {
 124             refreshConfigTask.dispose();
 125             refreshConfigTask = null;
 126         }
 127     }
 128
 129     public synchronized ConsumerConfiguration getDmaapConsumerConfiguration() {
 130         return dmaapConsumerConfiguration;
 131     }
 132
 133     /**
 134      * Checks if there is a configuration for the given feed.
 135      *
 136      * @param changeIdentifier the change identifier the feed is configured to belong to.
 137      * @return true if a feed is configured for the given change identifier, false if not.
 138      */
 139     public synchronized boolean isFeedConfigured(String changeIdentifier) {
 140         return publishingConfigurations.containsKey(changeIdentifier);
 141     }
 142
 143     /**
 144      * Gets the feed configuration for the given change identifier.
 145      *
 146      * @param changeIdentifier the change identifier the feed is configured to belong to.
 147      * @return the <code>PublisherConfiguration</code> for the feed belonging to the given change identifier.
 148      * @throws DatafileTaskException if no configuration has been loaded or the configuration is missing for the given
 149      *         change identifier.
 150      */
 151     public synchronized PublisherConfiguration getPublisherConfiguration(String changeIdentifier)
 152         throws DatafileTaskException {
 153
 154         if (publishingConfigurations == null) {
 155             throw new DatafileTaskException("No PublishingConfiguration loaded, changeIdentifier: " + changeIdentifier);
 156         }
 157         PublisherConfiguration cfg = publishingConfigurations.get(changeIdentifier);
 158         if (cfg == null) {
 159             throw new DatafileTaskException(
 160                 "Cannot find getPublishingConfiguration for changeIdentifier: " + changeIdentifier);
 161         }
 162         return cfg;
 163     }
 164
 165     public synchronized FtpesConfig getFtpesConfiguration() {
 166         return ftpesConfiguration;
 167     }
 168
 169     public synchronized SftpConfig getSftpConfiguration() {
 170         return sftpConfiguration;
 171     }
 172
 173     private <R> Mono<R> onErrorResume(Throwable trowable) {
 174         logger.error("Could not refresh application configuration {}", trowable.toString());
 175         return Mono.empty();
 176     }
 177
 178     Mono<EnvProperties> getEnvironment(Properties systemEnvironment, Map<String, String> context) {
 179         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context);
 180     }
 181
 182     Mono<CbsClient> createCbsClient(EnvProperties env) {
 183         return CbsClientFactory.createCbsClient(env);
 184     }
 185
 186     private AppConfig parseCloudConfig(JsonObject configurationObject) {
 187         try {
 188             CloudConfigParser parser = new CloudConfigParser(configurationObject, systemEnvironment);
 189             setConfiguration(parser.getDmaapConsumerConfig(), parser.getDmaapPublisherConfigurations(),
 190                 parser.getFtpesConfig(), parser.getSftpConfig());
 191             logConfig();
 192         } catch (DatafileTaskException e) {
 193             logger.error("Could not parse configuration {}", e.toString(), e);
 194         }
 195         return this;
 196     }
 197
 198     private void logConfig() {
 199         logger.debug("Read and parsed sFTP configuration:      [{}]", sftpConfiguration);
 200         logger.debug("Read and parsed FTPes configuration:     [{}]", ftpesConfiguration);
 201         logger.debug("Read and parsed DMaaP configuration:     [{}]", dmaapConsumerConfiguration);
 202         logger.debug("Read and parsed Publish configuration:   [{}]", publishingConfigurations);
 203     }
 204
 205     void loadConfigurationFromFile() {
 206         GsonBuilder gsonBuilder = new GsonBuilder();
 207         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 208
 209         try (InputStream inputStream = createInputStream(filepath)) {
 210             JsonParser parser = new JsonParser();
 211             JsonObject rootObject = getJsonElement(parser, inputStream).getAsJsonObject();
 212             if (rootObject == null) {
 213                 throw new JsonSyntaxException("Root is not a json object");
 214             }
 215             parseCloudConfig(rootObject);
 216             logger.info("Local configuration file loaded: {}", filepath);
 217         } catch (JsonSyntaxException | IOException e) {
 218             logger.trace("Local configuration file not loaded: {}", filepath, e);
 219         }
 220     }
 221
 222     private synchronized void setConfiguration(@NotNull ConsumerConfiguration consumerConfiguration,
 223         @NotNull Map<String, PublisherConfiguration> publisherConfiguration, @NotNull FtpesConfig ftpesConfig,
 224         @NotNull SftpConfig sftpConfig) {
 225         this.dmaapConsumerConfiguration = consumerConfiguration;
 226         this.publishingConfigurations = publisherConfiguration;
 227         this.ftpesConfiguration = ftpesConfig;
 228         this.sftpConfiguration = sftpConfig;
 229     }
 230
 231     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
 232         return parser.parse(new InputStreamReader(inputStream));
 233     }
 234
 235     InputStream createInputStream(@NotNull String filepath) throws IOException {
 236         return new BufferedInputStream(new FileInputStream(filepath));
 237     }
 238
 239 }