2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
6 * Copyright © 2017 Amdocs
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
12 * http://www.apache.org/licenses/LICENSE-2.0
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 * ============LICENSE_END=========================================================
21 * ECOMP is a trademark and service mark of AT&T Intellectual Property.
23 package org.onap.aai.sparky.synchronizer;
25 import com.google.common.util.concurrent.ThreadFactoryBuilder;
27 import java.lang.Thread.UncaughtExceptionHandler;
28 import java.text.SimpleDateFormat;
29 import java.util.ArrayList;
30 import java.util.Calendar;
31 import java.util.List;
33 import java.util.TimeZone;
34 import java.util.concurrent.Executors;
35 import java.util.concurrent.ScheduledExecutorService;
36 import java.util.concurrent.ThreadFactory;
37 import java.util.concurrent.TimeUnit;
38 import java.util.concurrent.atomic.AtomicLong;
40 import org.onap.aai.sparky.config.oxm.OxmEntityDescriptor;
41 import org.onap.aai.sparky.config.oxm.OxmModelLoader;
42 import org.onap.aai.sparky.dal.aai.ActiveInventoryAdapter;
43 import org.onap.aai.sparky.dal.aai.config.ActiveInventoryConfig;
44 import org.onap.aai.sparky.dal.aai.config.ActiveInventoryRestConfig;
45 import org.onap.aai.sparky.dal.cache.EntityCache;
46 import org.onap.aai.sparky.dal.cache.InMemoryEntityCache;
47 import org.onap.aai.sparky.dal.cache.PersistentEntityCache;
48 import org.onap.aai.sparky.dal.elasticsearch.ElasticSearchAdapter;
49 import org.onap.aai.sparky.dal.elasticsearch.config.ElasticSearchConfig;
50 import org.onap.aai.sparky.dal.rest.RestClientBuilder;
51 import org.onap.aai.sparky.dal.rest.RestfulDataAccessor;
52 import org.onap.aai.sparky.logging.AaiUiMsgs;
53 import org.onap.aai.sparky.synchronizer.SyncController.SyncActions;
54 import org.onap.aai.sparky.synchronizer.config.SynchronizerConfiguration;
55 import org.onap.aai.sparky.synchronizer.config.SynchronizerConstants;
56 import org.onap.aai.sparky.synchronizer.enumeration.SynchronizerState;
57 import org.onap.aai.sparky.util.ErrorUtil;
58 import org.onap.aai.sparky.viewandinspect.config.TierSupportUiConstants;
59 import org.onap.aai.cl.api.Logger;
60 import org.onap.aai.cl.eelf.LoggerFactory;
64 * The Class SyncHelper.
68 public class SyncHelper {
70 private final Logger LOG = LoggerFactory.getInstance().getLogger(SyncHelper.class);
71 private SyncController syncController = null;
72 private SyncController entityCounterHistorySummarizer = null;
74 private ScheduledExecutorService oneShotExecutor = Executors.newSingleThreadScheduledExecutor();
75 private ScheduledExecutorService periodicExecutor = null;
76 private ScheduledExecutorService historicalExecutor =
77 Executors.newSingleThreadScheduledExecutor();
79 private SynchronizerConfiguration syncConfig;
80 private ElasticSearchConfig esConfig;
81 private OxmModelLoader oxmModelLoader;
83 private Boolean initialSyncRunning = false;
84 private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss z");
85 private AtomicLong timeNextSync = new AtomicLong();
86 Map<String, String> contextMap;
91 private class SyncTask implements Runnable {
93 private boolean isInitialSync;
96 * Instantiates a new sync task.
98 * @param initialSync the initial sync
100 public SyncTask(boolean initialSync) {
101 this.isInitialSync = initialSync;
107 * @see java.lang.Runnable#run()
111 long opStartTime = System.currentTimeMillis();
112 MDC.setContextMap(contextMap);
114 LOG.info(AaiUiMsgs.SEARCH_ENGINE_SYNC_STARTED, sdf.format(opStartTime)
115 .replaceAll(SynchronizerConstants.TIME_STD, SynchronizerConstants.TIME_CONFIG_STD));
119 if (syncController == null) {
120 LOG.error(AaiUiMsgs.SYNC_SKIPPED_SYNCCONTROLLER_NOT_INITIALIZED);
124 int taskFrequencyInDays = SynchronizerConfiguration.getConfig().getSyncTaskFrequencyInDay();
127 * Do nothing if the initial start-up sync hasn't finished yet, but the regular sync
128 * scheduler fired up a regular sync.
130 if (!initialSyncRunning) {
132 initialSyncRunning = true;
134 // update 'timeNextSync' for periodic sync
135 timeNextSync.getAndAdd(taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);
139 LOG.info(AaiUiMsgs.INFO_GENERIC, "SyncTask, starting syncrhonization");
141 syncController.performAction(SyncActions.SYNCHRONIZE);
143 while (syncController.getState() == SynchronizerState.PERFORMING_SYNCHRONIZATION) {
148 LOG.info(AaiUiMsgs.SKIP_PERIODIC_SYNC_AS_SYNC_DIDNT_FINISH, sdf.format(opStartTime)
149 .replaceAll(SynchronizerConstants.TIME_STD, SynchronizerConstants.TIME_CONFIG_STD));
154 long opEndTime = System.currentTimeMillis();
158 * Handle corner case when start-up sync operation overlapped with a scheduled
159 * sync-start-time. Note that the scheduled sync does nothing if 'initialSyncRunning' is
160 * TRUE. So the actual next-sync is one more sync-cycle away
162 long knownNextSyncTime = timeNextSync.get();
163 if (knownNextSyncTime != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS
164 && opEndTime > knownNextSyncTime) {
165 timeNextSync.compareAndSet(knownNextSyncTime,
166 knownNextSyncTime + taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);
167 initialSyncRunning = false;
171 String durationMessage =
172 String.format(syncController.getControllerName() + " synchronization took '%d' ms.",
173 (opEndTime - opStartTime));
175 LOG.info(AaiUiMsgs.SYNC_DURATION, durationMessage);
177 // Provide log about the time for next synchronization
178 if (syncConfig.isConfigOkForPeriodicSync()
179 && timeNextSync.get() != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
180 TimeZone tz = TimeZone.getTimeZone(syncConfig.getSyncTaskStartTimeTimeZone());
182 if (opEndTime - opStartTime > taskFrequencyInDays
183 * SynchronizerConstants.MILLISEC_IN_A_DAY) {
184 String durationWasLongerMessage = String.format(
185 syncController.getControllerName()
186 + " synchronization took '%d' ms which is larger than"
187 + " synchronization interval of '%d' ms.",
188 (opEndTime - opStartTime),
189 taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);
191 LOG.info(AaiUiMsgs.SYNC_DURATION, durationWasLongerMessage);
194 LOG.info(AaiUiMsgs.SYNC_TO_BEGIN, syncController.getControllerName(),
195 sdf.format(timeNextSync).replaceAll(SynchronizerConstants.TIME_STD,
196 SynchronizerConstants.TIME_CONFIG_STD));
199 } catch (Exception exc) {
200 String message = "Caught an exception while attempt to synchronize elastic search "
201 + "with an error cause = " + ErrorUtil.extractStackTraceElements(5, exc);
202 LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
211 * Gets the first sync time.
213 * @param calendar the calendar
214 * @param timeNow the time now
215 * @param taskFreqInDay the task freq in day
216 * @return the first sync time
218 public long getFirstSyncTime(Calendar calendar, long timeNow, int taskFreqInDay) {
219 if (taskFreqInDay == SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
220 return SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS;
221 } else if (timeNow > calendar.getTimeInMillis()) {
222 calendar.add(Calendar.DAY_OF_MONTH, taskFreqInDay);
224 return calendar.getTimeInMillis();
228 * Boot strap and configure the moving pieces of the Sync Controller.
231 private void initializeSyncController() {
236 * TODO: it would be nice to have XML IoC / dependency injection kind of thing for these
237 * pieces maybe Spring?
241 * Sync Controller itself
244 syncController = new SyncController("entitySyncController");
247 * Create common elements
250 ActiveInventoryAdapter aaiAdapter = new ActiveInventoryAdapter(new RestClientBuilder());
251 ActiveInventoryRestConfig aaiRestConfig =
252 ActiveInventoryConfig.getConfig().getAaiRestConfig();
255 EntityCache cache = null;
257 if (aaiRestConfig.isCacheEnabled()) {
258 cache = new PersistentEntityCache(aaiRestConfig.getStorageFolderOverride(),
259 aaiRestConfig.getNumCacheWorkers());
261 cache = new InMemoryEntityCache();
264 RestClientBuilder clientBuilder = new RestClientBuilder();
266 aaiAdapter.setCacheEnabled(true);
267 aaiAdapter.setEntityCache(cache);
269 clientBuilder.setUseHttps(false);
271 RestfulDataAccessor nonCachingRestProvider = new RestfulDataAccessor(clientBuilder);
273 ElasticSearchConfig esConfig = ElasticSearchConfig.getConfig();
274 ElasticSearchAdapter esAdapter = new ElasticSearchAdapter(nonCachingRestProvider, esConfig);
277 * Register Index Validators
280 IndexIntegrityValidator entitySearchIndexValidator =
281 new IndexIntegrityValidator(nonCachingRestProvider, esConfig.getIndexName(),
282 esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
283 esConfig.buildElasticSearchTableConfig());
285 syncController.registerIndexValidator(entitySearchIndexValidator);
287 // TODO: Insert IndexValidator for TopographicalEntityIndex
288 // we should have one, but one isn't 100% required as none of the fields are analyzed
291 * Register Synchronizers
294 SearchableEntitySynchronizer ses = new SearchableEntitySynchronizer(esConfig.getIndexName());
295 ses.setAaiDataProvider(aaiAdapter);
296 ses.setEsDataProvider(esAdapter);
297 syncController.registerEntitySynchronizer(ses);
299 CrossEntityReferenceSynchronizer cers = new CrossEntityReferenceSynchronizer(
300 esConfig.getIndexName(), ActiveInventoryConfig.getConfig());
301 cers.setAaiDataProvider(aaiAdapter);
302 cers.setEsDataProvider(esAdapter);
303 syncController.registerEntitySynchronizer(cers);
305 if (syncConfig.isAutosuggestSynchronizationEnabled()) {
306 initAutoSuggestionSynchronizer(esConfig, aaiAdapter, esAdapter, nonCachingRestProvider);
307 initAggregationSynchronizer(esConfig, aaiAdapter, esAdapter, nonCachingRestProvider);
314 IndexCleaner searchableIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
315 esConfig.getIndexName(), esConfig.getType(), esConfig.getIpAddress(),
316 esConfig.getHttpPort(), syncConfig.getScrollContextTimeToLiveInMinutes(),
317 syncConfig.getNumScrollContextItemsToRetrievePerRequest());
319 syncController.registerIndexCleaner(searchableIndexCleaner);
321 } catch (Exception exc) {
322 String message = "Error: failed to sync with message = " + exc.getMessage();
323 LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
328 private List<String> getAutosuggestableEntitiesFromOXM() {
329 Map<String, OxmEntityDescriptor> map = oxmModelLoader.getSuggestionSearchEntityDescriptors();
330 List<String> suggestableEntities = new ArrayList<String>();
332 for (String entity: map.keySet()){
333 suggestableEntities.add(entity);
335 return suggestableEntities;
339 * Initialize the AutosuggestionSynchronizer and
340 * AggregationSuggestionSynchronizer
345 * @param nonCachingRestProvider
347 private void initAutoSuggestionSynchronizer(ElasticSearchConfig esConfig,
348 ActiveInventoryAdapter aaiAdapter, ElasticSearchAdapter esAdapter,
349 RestfulDataAccessor nonCachingRestProvider) {
350 LOG.info(AaiUiMsgs.INFO_GENERIC, "initAutoSuggestionSynchronizer");
352 // Initialize for entityautosuggestindex
354 IndexIntegrityValidator autoSuggestionIndexValidator =
355 new IndexIntegrityValidator(nonCachingRestProvider, esConfig.getAutosuggestIndexname(),
356 esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
357 esConfig.buildAutosuggestionTableConfig());
359 syncController.registerIndexValidator(autoSuggestionIndexValidator);
361 AutosuggestionSynchronizer suggestionSynchronizer =
362 new AutosuggestionSynchronizer(esConfig.getAutosuggestIndexname());
363 suggestionSynchronizer.setAaiDataProvider(aaiAdapter);
364 suggestionSynchronizer.setEsDataProvider(esAdapter);
365 syncController.registerEntitySynchronizer(suggestionSynchronizer);
367 AggregationSuggestionSynchronizer aggregationSuggestionSynchronizer =
368 new AggregationSuggestionSynchronizer(esConfig.getAutosuggestIndexname());
369 aggregationSuggestionSynchronizer.setEsDataProvider(esAdapter);
370 syncController.registerEntitySynchronizer(aggregationSuggestionSynchronizer);
372 IndexCleaner autosuggestIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
373 esConfig.getAutosuggestIndexname(), esConfig.getType(), esConfig.getIpAddress(),
374 esConfig.getHttpPort(), syncConfig.getScrollContextTimeToLiveInMinutes(),
375 syncConfig.getNumScrollContextItemsToRetrievePerRequest());
377 syncController.registerIndexCleaner(autosuggestIndexCleaner);
378 } catch (Exception exc) {
379 String message = "Error: failed to sync with message = " + exc.getMessage();
380 LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
385 * Initialize the AggregationSynchronizer
390 * @param nonCachingRestProvider
392 private void initAggregationSynchronizer(ElasticSearchConfig esConfig,
393 ActiveInventoryAdapter aaiAdapter, ElasticSearchAdapter esAdapter,
394 RestfulDataAccessor nonCachingRestProvider) {
395 LOG.info(AaiUiMsgs.INFO_GENERIC, "initAggregationSynchronizer");
397 List<String> aggregationEntities = getAutosuggestableEntitiesFromOXM();
399 // For each index: create an IndexValidator, a Synchronizer, and an IndexCleaner
400 for (String entity : aggregationEntities) {
402 String indexName = TierSupportUiConstants.getAggregationIndexName(entity);
404 IndexIntegrityValidator aggregationIndexValidator = new IndexIntegrityValidator(
405 nonCachingRestProvider, indexName, esConfig.getType(), esConfig.getIpAddress(),
406 esConfig.getHttpPort(), esConfig.buildAggregationTableConfig());
408 syncController.registerIndexValidator(aggregationIndexValidator);
411 * TODO: This per-entity-synchronizer approach will eventually result in AAI / ES overload
412 * because of the existing dedicated thread pools for ES + AAI operations within the
413 * synchronizer. If we had 50 types to sync then the thread pools within each Synchronizer
414 * would cause some heartburn as there would be hundreds of threads trying to talk to AAI.
415 * Given that we our running out of time, let's make sure we can get it functional and then
418 AggregationSynchronizer aggSynchronizer = new AggregationSynchronizer(entity, indexName);
419 aggSynchronizer.setAaiDataProvider(aaiAdapter);
420 aggSynchronizer.setEsDataProvider(esAdapter);
421 syncController.registerEntitySynchronizer(aggSynchronizer);
423 IndexCleaner entityDataIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
424 indexName, esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
425 syncConfig.getScrollContextTimeToLiveInMinutes(),
426 syncConfig.getNumScrollContextItemsToRetrievePerRequest());
428 syncController.registerIndexCleaner(entityDataIndexCleaner);
430 } catch (Exception exc) {
431 String message = "Error: failed to sync with message = " + exc.getMessage();
432 LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
438 * Instantiates a new sync helper.
440 * @param loader the loader
442 public SyncHelper(OxmModelLoader loader) {
444 this.contextMap = MDC.getCopyOfContextMap();
445 this.syncConfig = SynchronizerConfiguration.getConfig();
446 this.esConfig = ElasticSearchConfig.getConfig();
447 this.oxmModelLoader = loader;
449 UncaughtExceptionHandler uncaughtExceptionHandler = new Thread.UncaughtExceptionHandler() {
452 public void uncaughtException(Thread thread, Throwable exc) {
453 LOG.error(AaiUiMsgs.ERROR_GENERIC, thread.getName() + ": " + exc);
457 ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("SyncHelper-%d")
458 .setUncaughtExceptionHandler(uncaughtExceptionHandler).build();
460 periodicExecutor = Executors.newScheduledThreadPool(3, namedThreadFactory);
463 * We only want to initialize the synchronizer if sync has been configured to start
465 if (syncConfig.isConfigOkForStartupSync() || syncConfig.isConfigOkForPeriodicSync()) {
466 initializeSyncController();
469 // schedule startup synchronization
470 if (syncConfig.isConfigOkForStartupSync()) {
472 long taskInitialDelayInMs = syncConfig.getSyncTaskInitialDelayInMs();
473 if (taskInitialDelayInMs != SynchronizerConstants.DELAY_NO_STARTUP_SYNC_IN_MS) {
474 oneShotExecutor.schedule(new SyncTask(true), taskInitialDelayInMs, TimeUnit.MILLISECONDS);
475 LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine startup synchronization is enabled.");
477 LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine startup synchronization is disabled.");
481 // schedule periodic synchronization
482 if (syncConfig.isConfigOkForPeriodicSync()) {
484 TimeZone tz = TimeZone.getTimeZone(syncConfig.getSyncTaskStartTimeTimeZone());
485 Calendar calendar = Calendar.getInstance(tz);
488 calendar.set(Calendar.HOUR_OF_DAY, syncConfig.getSyncTaskStartTimeHr());
489 calendar.set(Calendar.MINUTE, syncConfig.getSyncTaskStartTimeMin());
490 calendar.set(Calendar.SECOND, syncConfig.getSyncTaskStartTimeSec());
492 long timeCurrent = calendar.getTimeInMillis();
493 int taskFrequencyInDay = syncConfig.getSyncTaskFrequencyInDay();
494 timeNextSync.getAndSet(getFirstSyncTime(calendar, timeCurrent, taskFrequencyInDay));
496 long delayUntilFirstRegSyncInMs = 0;
497 delayUntilFirstRegSyncInMs = timeNextSync.get() - timeCurrent;
499 // Do all calculation in milliseconds
500 long taskFreqencyInMs = taskFrequencyInDay * SynchronizerConstants.MILLISEC_IN_A_DAY;
502 if (taskFreqencyInMs != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
503 periodicExecutor.scheduleAtFixedRate(new SyncTask(false), delayUntilFirstRegSyncInMs,
504 taskFreqencyInMs, TimeUnit.MILLISECONDS);
505 LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine periodic synchronization is enabled.");
506 // case: when - startup sync is misconfigured or is disabled
507 // - give a clue to user when is the next periodic sync
508 if (!syncConfig.isConfigOkForStartupSync()
509 || syncConfig.isConfigDisabledForInitialSync()) {
510 LOG.info(AaiUiMsgs.SYNC_TO_BEGIN, syncController.getControllerName(),
511 sdf.format(timeNextSync).replaceAll(SynchronizerConstants.TIME_STD,
512 SynchronizerConstants.TIME_CONFIG_STD));
515 LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine periodic synchronization is disabled.");
519 } catch (Exception exc) {
520 String message = "Caught an exception while starting up the SyncHelper. Error cause = \n"
521 + ErrorUtil.extractStackTraceElements(5, exc);
522 LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
530 public void shutdown() {
532 if (oneShotExecutor != null) {
533 oneShotExecutor.shutdown();
536 if (periodicExecutor != null) {
537 periodicExecutor.shutdown();
540 if (historicalExecutor != null) {
541 historicalExecutor.shutdown();
544 if (syncController != null) {
545 syncController.shutdown();
548 if (entityCounterHistorySummarizer != null) {
549 entityCounterHistorySummarizer.shutdown();
554 public OxmModelLoader getOxmModelLoader() {
555 return oxmModelLoader;
558 public void setOxmModelLoader(OxmModelLoader oxmModelLoader) {
559 this.oxmModelLoader = oxmModelLoader;