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

   1 /*-
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018, 2020-2021 NOKIA Intellectual Property, 2018-2019 Nordix Foundation.
   4  * All rights reserved.
   5  * ===============================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   7  * in compliance with the License. You may obtain a copy of the License at
   8  *
   9  * http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software distributed under the License
  12  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  13  * or implied. See the License for the specific language governing permissions and limitations under
  14  * the License.
  15  * ============LICENSE_END========================================================================
  16  */
  17
  18 package org.onap.dcaegen2.collectors.datafile.configuration;
  19
  20 import com.google.gson.GsonBuilder;
  21 import com.google.gson.JsonElement;
  22 import com.google.gson.JsonObject;
  23 import com.google.gson.JsonParser;
  24 import com.google.gson.JsonSyntaxException;
  25 import com.google.gson.TypeAdapterFactory;
  26
  27 import java.io.BufferedInputStream;
  28 import java.io.FileInputStream;
  29 import java.io.IOException;
  30 import java.io.InputStream;
  31 import java.io.InputStreamReader;
  32 import java.time.Duration;
  33 import java.util.Map;
  34 import java.util.Properties;
  35 import java.util.ServiceLoader;
  36
  37 import javax.validation.constraints.NotEmpty;
  38 import javax.validation.constraints.NotNull;
  39
  40 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  41 import org.onap.dcaegen2.collectors.datafile.http.HttpsClientConnectionManagerUtil;
  42 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
  43 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
  44 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
  45 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.CbsClientConfigurationException;
  46 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
  47 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
  48 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51 import org.springframework.beans.factory.annotation.Value;
  52 import org.springframework.boot.context.properties.ConfigurationProperties;
  53 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  54 import org.springframework.context.annotation.ComponentScan;
  55 import org.springframework.stereotype.Component;
  56
  57 import reactor.core.Disposable;
  58 import reactor.core.publisher.Flux;
  59 import reactor.core.publisher.Mono;
  60
  61 /**
  62  * Holds all configuration for the DFC.
  63  *
  64  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  65  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  66  */
  67
  68 @Component
  69 @ComponentScan("org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers")
  70 @EnableConfigurationProperties
  71 @ConfigurationProperties("app")
  72 public class AppConfig {
  73
  74     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
  75
  76     @Value("#{systemEnvironment}")
  77     Properties systemEnvironment;
  78     private ConsumerConfiguration dmaapConsumerConfiguration;
  79     private Map<String, PublisherConfiguration> publishingConfigurations;
  80     private CertificateConfig certificateConfiguration;
  81     private SftpConfig sftpConfiguration;
  82     private Disposable refreshConfigTask = null;
  83
  84     @NotEmpty
  85     private String filepath;
  86
  87     public synchronized void setFilepath(String filepath) {
  88         this.filepath = filepath;
  89     }
  90
  91     /**
  92      * Reads the cloud configuration.
  93      */
  94     public void initialize() {
  95         stop();
  96
  97         loadConfigurationFromFile();
  98
  99         refreshConfigTask = createRefreshTask() //
 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() {
 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 CertificateConfig getCertificateConfiguration() {
 167         return certificateConfiguration;
 168     }
 169
 170     public synchronized SftpConfig getSftpConfiguration() {
 171         return sftpConfiguration;
 172     }
 173
 174     private <R> Mono<R> onErrorResume(Throwable throwable) {
 175         String throwableString = throwable.toString();
 176         logger.error("Could not refresh application configuration {}", throwableString);
 177         return Mono.empty();
 178     }
 179
 180     Mono<CbsClientConfiguration> createCbsClientConfiguration() {
 181         try {
 182             return Mono.just(CbsClientConfiguration.fromEnvironment());
 183         } catch (CbsClientConfigurationException e) {
 184             return Mono.error(e);
 185         }
 186     }
 187
 188     Mono<CbsClient> createCbsClient(CbsClientConfiguration cbsClientConfiguration) {
 189         return CbsClientFactory.createCbsClient(cbsClientConfiguration);
 190     }
 191
 192     private AppConfig parseCloudConfig(JsonObject configurationObject) {
 193         try {
 194             CloudConfigParser parser =
 195                 new CloudConfigParser(configurationObject, systemEnvironment);
 196             setConfiguration(parser.getConsumerConfiguration(),
 197                 parser.getDmaapPublisherConfigurations(), parser.getCertificateConfig(),
 198                 parser.getSftpConfig());
 199             logConfig();
 200         } catch (DatafileTaskException e) {
 201             logger.error("Could not parse configuration {}", e.toString(), e);
 202         }
 203         return this;
 204     }
 205
 206     private void logConfig() {
 207         logger.debug("Read and parsed sFTP configuration:      [{}]", sftpConfiguration);
 208         logger.debug("Read and parsed FTPes / HTTPS configuration:     [{}]", certificateConfiguration);
 209         logger.debug("Read and parsed DMaaP configuration:     [{}]", dmaapConsumerConfiguration);
 210         logger.debug("Read and parsed Publish configuration:   [{}]", publishingConfigurations);
 211     }
 212
 213     void loadConfigurationFromFile() {
 214         GsonBuilder gsonBuilder = new GsonBuilder();
 215         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
 216
 217         try (InputStream inputStream = createInputStream(filepath)) {
 218             JsonObject rootObject = getJsonElement(inputStream).getAsJsonObject();
 219             if (rootObject == null) {
 220                 throw new JsonSyntaxException("Root is not a json object");
 221             }
 222             parseCloudConfig(rootObject);
 223             logger.info("Local configuration file loaded: {}", filepath);
 224         } catch (JsonSyntaxException | IOException e) {
 225             logger.trace("Local configuration file not loaded: {}", filepath, e);
 226         }
 227     }
 228
 229     private synchronized void setConfiguration(@NotNull ConsumerConfiguration consumerConfiguration,
 230         @NotNull Map<String, PublisherConfiguration> publisherConfiguration, @NotNull CertificateConfig certificateConfig,
 231         @NotNull SftpConfig sftpConfig) throws DatafileTaskException {
 232         this.dmaapConsumerConfiguration = consumerConfiguration;
 233         this.publishingConfigurations = publisherConfiguration;
 234         this.certificateConfiguration = certificateConfig;
 235         this.sftpConfiguration = sftpConfig;
 236
 237         HttpsClientConnectionManagerUtil.setupOrUpdate(certificateConfig.keyCert(), certificateConfig.keyPasswordPath(),
 238             certificateConfig.trustedCa(), certificateConfig.trustedCaPasswordPath(),
 239             certificateConfig.httpsHostnameVerify());
 240     }
 241
 242     JsonElement getJsonElement(InputStream inputStream) {
 243         return JsonParser.parseReader(new InputStreamReader(inputStream));
 244     }
 245
 246     InputStream createInputStream(@NotNull String filepath) throws IOException {
 247         return new BufferedInputStream(new FileInputStream(filepath));
 248     }
 249
 250 }