Remove call stack from original exceptions in log
[nonrtric.git] / policy-agent / src / main / java / org / oransc / policyagent / tasks / RefreshConfigTask.java
1 /*-
2  * ========================LICENSE_START=================================
3  * O-RAN-SC
4  * %%
5  * Copyright (C) 2019 Nordix Foundation
6  * %%
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  * ========================LICENSE_END===================================
19  */
20
21 package org.oransc.policyagent.tasks;
22
23 import com.google.gson.GsonBuilder;
24 import com.google.gson.JsonElement;
25 import com.google.gson.JsonObject;
26 import com.google.gson.JsonParser;
27 import com.google.gson.JsonSyntaxException;
28 import com.google.gson.TypeAdapterFactory;
29
30 import java.io.BufferedInputStream;
31 import java.io.File;
32 import java.io.FileInputStream;
33 import java.io.IOException;
34 import java.io.InputStream;
35 import java.io.InputStreamReader;
36 import java.time.Duration;
37 import java.util.Properties;
38 import java.util.ServiceLoader;
39
40 import javax.validation.constraints.NotNull;
41
42 import lombok.AccessLevel;
43 import lombok.Getter;
44
45 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
46 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
47 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
48 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
49 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties;
50 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
51 import org.oransc.policyagent.clients.A1ClientFactory;
52 import org.oransc.policyagent.configuration.ApplicationConfig;
53 import org.oransc.policyagent.configuration.ApplicationConfig.RicConfigUpdate;
54 import org.oransc.policyagent.configuration.ApplicationConfigParser;
55 import org.oransc.policyagent.configuration.RicConfig;
56 import org.oransc.policyagent.exceptions.ServiceException;
57 import org.oransc.policyagent.repository.Policies;
58 import org.oransc.policyagent.repository.PolicyTypes;
59 import org.oransc.policyagent.repository.Ric;
60 import org.oransc.policyagent.repository.Rics;
61 import org.oransc.policyagent.repository.Services;
62 import org.slf4j.Logger;
63 import org.slf4j.LoggerFactory;
64 import org.springframework.beans.factory.annotation.Autowired;
65 import org.springframework.beans.factory.annotation.Value;
66 import org.springframework.stereotype.Component;
67
68 import reactor.core.Disposable;
69 import reactor.core.publisher.Flux;
70 import reactor.core.publisher.Mono;
71
72 /**
73  * Regularly refreshes the configuration from Consul or from a local
74  * configuration file.
75  */
76 @Component
77 public class RefreshConfigTask {
78
79     private static final Logger logger = LoggerFactory.getLogger(RefreshConfigTask.class);
80
81     @Value("#{systemEnvironment}")
82     public Properties systemEnvironment;
83
84     final ApplicationConfig appConfig;
85     @Getter(AccessLevel.PROTECTED)
86     private Disposable refreshTask = null;
87     private boolean isConsulUsed = false;
88
89     private final Rics rics;
90     private final A1ClientFactory a1ClientFactory;
91     private final Policies policies;
92     private final Services services;
93     private final PolicyTypes policyTypes;
94     private static final Duration FILE_CONFIG_REFRESH_INTERVAL = Duration.ofMinutes(1);
95     private static final Duration CONSUL_CONFIG_REFRESH_INTERVAL = Duration.ofMinutes(1);
96
97     @Autowired
98     public RefreshConfigTask(ApplicationConfig appConfig, Rics rics, Policies policies, Services services,
99         PolicyTypes policyTypes, A1ClientFactory a1ClientFactory) {
100         this.appConfig = appConfig;
101         this.rics = rics;
102         this.policies = policies;
103         this.services = services;
104         this.policyTypes = policyTypes;
105         this.a1ClientFactory = a1ClientFactory;
106     }
107
108     public void start() {
109         logger.debug("Starting refreshConfigTask");
110         stop();
111         refreshTask = createRefreshTask() //
112             .subscribe(
113                 notUsed -> logger.debug("Refreshed configuration data"), throwable -> logger
114                     .error("Configuration refresh terminated due to exception {}", throwable.getMessage()),
115                 () -> logger.error("Configuration refresh terminated"));
116     }
117
118     public void stop() {
119         if (refreshTask != null) {
120             refreshTask.dispose();
121         }
122     }
123
124     Flux<RicConfigUpdate.Type> createRefreshTask() {
125         Flux<JsonObject> loadFromFile = Flux.interval(Duration.ZERO, FILE_CONFIG_REFRESH_INTERVAL) //
126             .filter(notUsed -> configFileExists()) //
127             .filter(notUsed -> !this.isConsulUsed) //
128             .flatMap(notUsed -> loadConfigurationFromFile()) //
129             .onErrorResume(this::ignoreError) //
130             .doOnNext(json -> logger.debug("loadFromFile")) //
131             .doOnTerminate(() -> logger.error("loadFromFile Terminate"));
132
133         Flux<JsonObject> loadFromConsul = getEnvironment(systemEnvironment) //
134             .flatMap(this::createCbsClient) //
135             .flatMapMany(this::periodicConfigurationUpdates) //
136             .onErrorResume(this::ignoreError) //
137             .doOnNext(json -> logger.debug("loadFromConsul")) //
138             .doOnNext(json -> this.isConsulUsed = true) //
139             .doOnTerminate(() -> logger.error("loadFromConsul Terminate"));
140
141         return Flux.merge(loadFromFile, loadFromConsul) //
142             .flatMap(this::parseConfiguration) //
143             .flatMap(this::updateConfig) //
144             .doOnNext(this::handleUpdatedRicConfig) //
145             .flatMap(configUpdate -> Flux.just(configUpdate.getType())) //
146             .doOnTerminate(() -> logger.error("Configuration refresh task is terminated"));
147     }
148
149     Mono<EnvProperties> getEnvironment(Properties systemEnvironment) {
150         return EnvironmentProcessor.readEnvironmentVariables(systemEnvironment);
151     }
152
153     Mono<CbsClient> createCbsClient(EnvProperties env) {
154         return CbsClientFactory.createCbsClient(env);
155     }
156
157     private Flux<JsonObject> periodicConfigurationUpdates(CbsClient cbsClient) {
158         final Duration initialDelay = Duration.ZERO;
159         final CbsRequest getConfigRequest = CbsRequests.getAll(RequestDiagnosticContext.create());
160         return cbsClient.updates(getConfigRequest, initialDelay, CONSUL_CONFIG_REFRESH_INTERVAL);
161     }
162
163     private <R> Mono<R> ignoreError(Throwable throwable) {
164         String errMsg = throwable.toString();
165         logger.warn("Could not refresh application configuration. {}", errMsg);
166         return Mono.empty();
167     }
168
169     private Mono<ApplicationConfigParser.ConfigParserResult> parseConfiguration(JsonObject jsonObject) {
170         try {
171             ApplicationConfigParser parser = new ApplicationConfigParser();
172             return Mono.just(parser.parse(jsonObject));
173         } catch (ServiceException e) {
174             logger.error("Could not parse configuration {}", e.toString(), e);
175             return Mono.error(e);
176         }
177     }
178
179     private Flux<RicConfigUpdate> updateConfig(ApplicationConfigParser.ConfigParserResult config) {
180         return this.appConfig.setConfiguration(config);
181     }
182
183     boolean configFileExists() {
184         String filepath = appConfig.getLocalConfigurationFilePath();
185         return (filepath != null && (new File(filepath).exists()));
186     }
187
188     private void handleUpdatedRicConfig(RicConfigUpdate updatedInfo) {
189         synchronized (this.rics) {
190             String ricName = updatedInfo.getRicConfig().name();
191             RicConfigUpdate.Type event = updatedInfo.getType();
192             if (event == RicConfigUpdate.Type.ADDED) {
193                 addRic(updatedInfo.getRicConfig());
194             } else if (event == RicConfigUpdate.Type.REMOVED) {
195                 rics.remove(ricName);
196                 this.policies.removePoliciesForRic(ricName);
197             } else if (event == RicConfigUpdate.Type.CHANGED) {
198                 Ric ric = this.rics.get(ricName);
199                 if (ric == null) {
200                     // Should not happen,just for robustness
201                     addRic(updatedInfo.getRicConfig());
202                 } else {
203                     ric.setRicConfig(updatedInfo.getRicConfig());
204                 }
205             }
206         }
207     }
208
209     private void addRic(RicConfig config) {
210         Ric ric = new Ric(config);
211         this.rics.put(ric);
212         runRicSynchronization(ric);
213     }
214
215     void runRicSynchronization(Ric ric) {
216         RicSynchronizationTask synchronizationTask =
217             new RicSynchronizationTask(a1ClientFactory, policyTypes, policies, services);
218         synchronizationTask.run(ric);
219     }
220
221     /**
222      * Reads the configuration from file.
223      */
224     Flux<JsonObject> loadConfigurationFromFile() {
225         String filepath = appConfig.getLocalConfigurationFilePath();
226         GsonBuilder gsonBuilder = new GsonBuilder();
227         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
228
229         try (InputStream inputStream = createInputStream(filepath)) {
230             JsonObject rootObject = getJsonElement(inputStream).getAsJsonObject();
231             ApplicationConfigParser appParser = new ApplicationConfigParser();
232             appParser.parse(rootObject);
233             logger.debug("Local configuration file loaded: {}", filepath);
234             return Flux.just(rootObject);
235         } catch (JsonSyntaxException | ServiceException | IOException e) {
236             logger.debug("Local configuration file not loaded: {}", filepath, e);
237             return Flux.empty();
238         }
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 }