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     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
  72
  73     private ConsumerConfiguration dmaapConsumerConfiguration;
  74     private Map<String, PublisherConfiguration> publishingConfigurations;
  75     private FtpesConfig ftpesConfiguration;
  76     @Value("#{systemEnvironment}")
  77     Properties systemEnvironment;
  78     private Disposable refreshConfigTask = null;
  79
  80     @NotEmpty
  81     private String filepath;
  82
  83     public synchronized void setFilepath(String filepath) {
  84         this.filepath = filepath;
  85     }
  86
  87     /**
  88      * Reads the cloud configuration.
  89      */
  90     public void initialize() {
  91         stop();
  92         Map<String, String> context = MappedDiagnosticContext.initializeTraceContext();
  93         loadConfigurationFromFile();
  94
  95         refreshConfigTask = createRefreshTask(context) //
  96             .subscribe(e -> logger.info("Refreshed configuration data"),
  97                 throwable -> logger.error("Configuration refresh terminated due to exception", throwable),
  98                 () -> logger.error("Configuration refresh terminated"));
  99     }
 100
 101     Flux<AppConfig> createRefreshTask(Map<String, String> context) {
 102         return getEnvironment(systemEnvironment, context).flatMap(this::createCbsClient)
 103             .flatMapMany(this::periodicConfigurationUpdates).map(this::parseCloudConfig)
 104             .onErrorResume(this::onErrorResume);
 105     }
 106
 107     private Flux<JsonObject> periodicConfigurationUpdates(CbsClient cbsClient) {
 108         final Duration initialDelay = Duration.ZERO;
 109         final Duration refreshPeriod = Duration.ofMinutes(1);
 110         final CbsRequest getConfigRequest = CbsRequests.getAll(RequestDiagnosticContext.create());
 111         return cbsClient.updates(getConfigRequest, initialDelay, refreshPeriod);
 112     }
 113
 114     /**
 115      * Stops the refreshing of the configuration.
 116      */
 117     public void stop() {
 118         if (refreshConfigTask != null) {
 119             refreshConfigTask.dispose();
 120             refreshConfigTask = null;
 121         }
 122     }
 123
 124     public synchronized ConsumerConfiguration getDmaapConsumerConfiguration() {
 125         return dmaapConsumerConfiguration;
 126     }
 127
 128     /**
 129      * Checks if there is a configuration for the given feed.
 130      *
 131      * @param changeIdentifier the change identifier the feed is configured to belong to.
 132      *
 133      * @return true if a feed is configured for the given change identifier, false if not.
 134      */
 135     public synchronized boolean isFeedConfigured(String changeIdentifier) {
 136         return publishingConfigurations.containsKey(changeIdentifier);
 137     }
 138
 139     /**
 140      * Gets the feed configuration for the given change identifier.
 141      *
 142      * @param changeIdentifier the change identifier the feed is configured to belong to.
 143      * @return the <code>PublisherConfiguration</code> for the feed belonging to the given change identifier.
 144      *
 145      * @throws DatafileTaskException if no configuration has been loaded or the configuration is missing for the given
 146      *         change identifier.
 147      */
 148     public synchronized PublisherConfiguration getPublisherConfiguration(String changeIdentifier)
 149         throws DatafileTaskException {
 150
 151         if (publishingConfigurations == null) {
 152             throw new DatafileTaskException("No PublishingConfiguration loaded, changeIdentifier: " + changeIdentifier);
 153         }
 154         PublisherConfiguration cfg = publishingConfigurations.get(changeIdentifier);
 155         if (cfg == null) {
 156             throw new DatafileTaskException(
 157                 "Cannot find getPublishingConfiguration for changeIdentifier: " + changeIdentifier);
 158         }
 159         return cfg;
 160     }
 161
 162     public synchronized FtpesConfig getFtpesConfiguration() {
 163         return ftpesConfiguration;
 164     }
 165
 166     private <R> Mono<R> onErrorResume(Throwable trowable) {
 167         logger.error("Could not refresh application configuration {}", trowable.toString());
 168         return Mono.empty();
 169     }
 170
 171     Mono<EnvProperties> getEnvironment(Properties systemEnvironment, Map<String, String> context) {
 172         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment, context);
 173     }
 174
 175     Mono<CbsClient> createCbsClient(EnvProperties env) {
 176         return CbsClientFactory.createCbsClient(env);
 177     }
 178
 179     /**
 180      * Parse configuration.
 181      *
 182      * @param jsonObject the DFC service's configuration
 183      * @return this which is updated if successful
 184      */
 185     private AppConfig parseCloudConfig(JsonObject jsonObject) {
 186         try {
 187             CloudConfigParser parser = new CloudConfigParser(jsonObject);
 188             setConfiguration(parser.getDmaapConsumerConfig(), parser.getDmaapPublisherConfigurations(),
 189                 parser.getFtpesConfig());
 190
 191         } catch (DatafileTaskException e) {
 192             logger.error("Could not parse configuration {}", e.toString(), e);
 193         }
 194         return this;
 195     }
 196
 197     /**
 198      * Reads the configuration from file.
 199      */
 200     void loadConfigurationFromFile() {
 201         GsonBuilder gsonBuilder = new GsonBuilder();
 202         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 203
 204         try (InputStream inputStream = createInputStream(filepath)) {
 205             JsonParser parser = new JsonParser();
 206             JsonObject rootObject = getJsonElement(parser, inputStream).getAsJsonObject();
 207             if (rootObject == null) {
 208                 throw new JsonSyntaxException("Root is not a json object");
 209             }
 210             parseCloudConfig(rootObject);
 211         } catch (JsonSyntaxException | IOException e) {
 212             logger.warn("Local configuration file not loaded: {}", filepath, e);
 213         }
 214     }
 215
 216     private synchronized void setConfiguration(ConsumerConfiguration consumerConfiguration,
 217         Map<String, PublisherConfiguration> publisherConfiguration, FtpesConfig ftpesConfig) {
 218         if (consumerConfiguration == null || publisherConfiguration == null || ftpesConfig == null) {
 219             logger.error(
 220                 "Problem with configuration consumerConfiguration: {}, publisherConfiguration: {}, ftpesConfig: {}",
 221                 consumerConfiguration, publisherConfiguration, ftpesConfig);
 222         } else {
 223             this.dmaapConsumerConfiguration = consumerConfiguration;
 224             this.publishingConfigurations = publisherConfiguration;
 225             this.ftpesConfiguration = ftpesConfig;
 226         }
 227     }
 228
 229     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
 230         return parser.parse(new InputStreamReader(inputStream));
 231     }
 232
 233     InputStream createInputStream(@NotNull String filepath) throws IOException {
 234         return new BufferedInputStream(new FileInputStream(filepath));
 235     }
 236
 237 }