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