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

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