path: root/src/main/java/org/onap/aai/sparky/synchronizer/SyncHelper.java

/**
 * ============LICENSE_START=======================================================
 * org.onap.aai
 * ================================================================================
 * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
 * Copyright © 2017 Amdocs
 * ================================================================================
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * ============LICENSE_END=========================================================
 *
 * ECOMP is a trademark and service mark of AT&T Intellectual Property.
 */
package org.onap.aai.sparky.synchronizer;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.lang.Thread.UncaughtExceptionHandler;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.onap.aai.sparky.config.oxm.OxmEntityDescriptor;
import org.onap.aai.sparky.config.oxm.OxmModelLoader;
import org.onap.aai.sparky.dal.aai.ActiveInventoryAdapter;
import org.onap.aai.sparky.dal.aai.config.ActiveInventoryConfig;
import org.onap.aai.sparky.dal.aai.config.ActiveInventoryRestConfig;
import org.onap.aai.sparky.dal.cache.EntityCache;
import org.onap.aai.sparky.dal.cache.InMemoryEntityCache;
import org.onap.aai.sparky.dal.cache.PersistentEntityCache;
import org.onap.aai.sparky.dal.elasticsearch.ElasticSearchAdapter;
import org.onap.aai.sparky.dal.elasticsearch.config.ElasticSearchConfig;
import org.onap.aai.sparky.dal.rest.RestClientBuilder;
import org.onap.aai.sparky.dal.rest.RestfulDataAccessor;
import org.onap.aai.sparky.logging.AaiUiMsgs;
import org.onap.aai.sparky.synchronizer.SyncController.SyncActions;
import org.onap.aai.sparky.synchronizer.config.SynchronizerConfiguration;
import org.onap.aai.sparky.synchronizer.config.SynchronizerConstants;
import org.onap.aai.sparky.synchronizer.enumeration.SynchronizerState;
import org.onap.aai.sparky.util.ErrorUtil;
import org.onap.aai.sparky.viewandinspect.config.TierSupportUiConstants;
import org.onap.aai.cl.api.Logger;
import org.onap.aai.cl.eelf.LoggerFactory;
import org.slf4j.MDC;

/**
 * The Class SyncHelper.
 *
 * @author davea.
 */
public class SyncHelper {

  private final Logger LOG = LoggerFactory.getInstance().getLogger(SyncHelper.class);
  private SyncController syncController = null;
  private SyncController entityCounterHistorySummarizer = null;

  private ScheduledExecutorService oneShotExecutor = Executors.newSingleThreadScheduledExecutor();
  private ScheduledExecutorService periodicExecutor = null;
  private ScheduledExecutorService historicalExecutor =
      Executors.newSingleThreadScheduledExecutor();

  private SynchronizerConfiguration syncConfig;
  private ElasticSearchConfig esConfig;
  private OxmModelLoader oxmModelLoader;

  private Boolean initialSyncRunning = false;
  private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss z");
  private AtomicLong timeNextSync = new AtomicLong();
  Map<String, String> contextMap;

  /**
   * The Class SyncTask.
   */
  private class SyncTask implements Runnable {

    private boolean isInitialSync;

    public boolean isInitialSync() {
      return isInitialSync;
    }

    public void setInitialSync(boolean isInitialSync) {
      this.isInitialSync = isInitialSync;
    }

    /**
     * Instantiates a new sync task.
     *
     * @param initialSync the initial sync
     */
    public SyncTask(boolean initialSync) {
      this.isInitialSync = initialSync;
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
      long opStartTime = System.currentTimeMillis();
      MDC.setContextMap(contextMap);

      LOG.info(AaiUiMsgs.SEARCH_ENGINE_SYNC_STARTED, sdf.format(opStartTime)
          .replaceAll(SynchronizerConstants.TIME_STD, SynchronizerConstants.TIME_CONFIG_STD));

      try {

        if (syncController == null) {
          LOG.error(AaiUiMsgs.SYNC_SKIPPED_SYNCCONTROLLER_NOT_INITIALIZED);
          return;
        }

        int taskFrequencyInDays = SynchronizerConfiguration.getConfig().getSyncTaskFrequencyInDay();

        /*
         * Do nothing if the initial start-up sync hasn't finished yet, but the regular sync
         * scheduler fired up a regular sync.
         */
        if (!initialSyncRunning) {
          if (isInitialSync) {
            initialSyncRunning = true;
          } else {
            // update 'timeNextSync' for periodic sync
            timeNextSync.getAndAdd(taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);

          }

          LOG.info(AaiUiMsgs.INFO_GENERIC, "SyncTask, starting syncrhonization");

          syncController.performAction(SyncActions.SYNCHRONIZE);

          while (syncController.getState() == SynchronizerState.PERFORMING_SYNCHRONIZATION) {
            Thread.sleep(1000);
          }

        } else {
          LOG.info(AaiUiMsgs.SKIP_PERIODIC_SYNC_AS_SYNC_DIDNT_FINISH, sdf.format(opStartTime)
              .replaceAll(SynchronizerConstants.TIME_STD, SynchronizerConstants.TIME_CONFIG_STD));

          return;
        }

        long opEndTime = System.currentTimeMillis();

        if (isInitialSync) {
          /*
           * Handle corner case when start-up sync operation overlapped with a scheduled
           * sync-start-time. Note that the scheduled sync does nothing if 'initialSyncRunning' is
           * TRUE. So the actual next-sync is one more sync-cycle away
           */
          long knownNextSyncTime = timeNextSync.get();
          if (knownNextSyncTime != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS
              && opEndTime > knownNextSyncTime) {
            timeNextSync.compareAndSet(knownNextSyncTime,
                knownNextSyncTime + taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);
            initialSyncRunning = false;
          }
        }

        String durationMessage =
            String.format(syncController.getControllerName() + " synchronization took '%d' ms.",
                (opEndTime - opStartTime));

        LOG.info(AaiUiMsgs.SYNC_DURATION, durationMessage);

        // Provide log about the time for next synchronization
        if (syncConfig.isConfigOkForPeriodicSync()
            && timeNextSync.get() != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
          TimeZone tz = TimeZone.getTimeZone(syncConfig.getSyncTaskStartTimeTimeZone());
          sdf.setTimeZone(tz);
          if (opEndTime - opStartTime > taskFrequencyInDays
              * SynchronizerConstants.MILLISEC_IN_A_DAY) {
            String durationWasLongerMessage = String.format(
                syncController.getControllerName()
                    + " synchronization took '%d' ms which is larger than"
                    + " synchronization interval of '%d' ms.",
                (opEndTime - opStartTime),
                taskFrequencyInDays * SynchronizerConstants.MILLISEC_IN_A_DAY);

            LOG.info(AaiUiMsgs.SYNC_DURATION, durationWasLongerMessage);
          }

          LOG.info(AaiUiMsgs.SYNC_TO_BEGIN, syncController.getControllerName(),
              sdf.format(timeNextSync).replaceAll(SynchronizerConstants.TIME_STD,
                  SynchronizerConstants.TIME_CONFIG_STD));
        }

      } catch (Exception exc) {
        String message = "Caught an exception while attempt to synchronize elastic search "
            + "with an error cause = " + ErrorUtil.extractStackTraceElements(5, exc);
        LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
      }

    }

  }


  /**
   * Gets the first sync time.
   *
   * @param calendar the calendar
   * @param timeNow the time now
   * @param taskFreqInDay the task freq in day
   * @return the first sync time
   */
  public long getFirstSyncTime(Calendar calendar, long timeNow, int taskFreqInDay) {
    if (taskFreqInDay == SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
      return SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS;
    } else if (timeNow > calendar.getTimeInMillis()) {
      calendar.add(Calendar.DAY_OF_MONTH, taskFreqInDay);
    }
    return calendar.getTimeInMillis();
  }

  /**
   * Boot strap and configure the moving pieces of the Sync Controller.
   */

  private void initializeSyncController() {

    try {

      /*
       * TODO: it would be nice to have XML IoC / dependency injection kind of thing for these
       * pieces maybe Spring?
       */

      /*
       * Sync Controller itself
       */

      syncController = new SyncController("entitySyncController");

      /*
       * Create common elements
       */

      ActiveInventoryAdapter aaiAdapter = new ActiveInventoryAdapter(new RestClientBuilder());
      ActiveInventoryRestConfig aaiRestConfig =
          ActiveInventoryConfig.getConfig().getAaiRestConfig();


      EntityCache cache = null;

      if (aaiRestConfig.isCacheEnabled()) {
        cache = new PersistentEntityCache(aaiRestConfig.getStorageFolderOverride(),
            aaiRestConfig.getNumCacheWorkers());
      } else {
        cache = new InMemoryEntityCache();
      }

      RestClientBuilder clientBuilder = new RestClientBuilder();

      aaiAdapter.setCacheEnabled(true);
      aaiAdapter.setEntityCache(cache);

      clientBuilder.setUseHttps(false);

      RestfulDataAccessor nonCachingRestProvider = new RestfulDataAccessor(clientBuilder);

      ElasticSearchConfig esConfig = ElasticSearchConfig.getConfig();
      ElasticSearchAdapter esAdapter = new ElasticSearchAdapter(nonCachingRestProvider, esConfig);

      /*
       * Register Index Validators
       */

      IndexIntegrityValidator entitySearchIndexValidator =
          new IndexIntegrityValidator(nonCachingRestProvider, esConfig.getIndexName(),
              esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
              esConfig.buildElasticSearchTableConfig());

      syncController.registerIndexValidator(entitySearchIndexValidator);

      // TODO: Insert IndexValidator for TopographicalEntityIndex
      // we should have one, but one isn't 100% required as none of the fields are analyzed

      /*
       * Register Synchronizers
       */

      SearchableEntitySynchronizer ses = new SearchableEntitySynchronizer(esConfig.getIndexName());
      ses.setAaiDataProvider(aaiAdapter);
      ses.setEsDataProvider(esAdapter);
      syncController.registerEntitySynchronizer(ses);

      CrossEntityReferenceSynchronizer cers = new CrossEntityReferenceSynchronizer(
          esConfig.getIndexName(), ActiveInventoryConfig.getConfig());
      cers.setAaiDataProvider(aaiAdapter);
      cers.setEsDataProvider(esAdapter);
      syncController.registerEntitySynchronizer(cers);

      if (syncConfig.isAutosuggestSynchronizationEnabled()) {
        initAutoSuggestionSynchronizer(esConfig, aaiAdapter, esAdapter, nonCachingRestProvider);
        initAggregationSynchronizer(esConfig, aaiAdapter, esAdapter, nonCachingRestProvider);
      }

      /*
       * Register Cleaners
       */

      IndexCleaner searchableIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
          esConfig.getIndexName(), esConfig.getType(), esConfig.getIpAddress(),
          esConfig.getHttpPort(), syncConfig.getScrollContextTimeToLiveInMinutes(),
          syncConfig.getNumScrollContextItemsToRetrievePerRequest());

      syncController.registerIndexCleaner(searchableIndexCleaner);

    } catch (Exception exc) {
      String message = "Error: failed to sync with message = " + exc.getMessage();
      LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
    }

  }

  private List<String> getAutosuggestableEntitiesFromOXM() {
    Map<String, OxmEntityDescriptor> map = oxmModelLoader.getSuggestionSearchEntityDescriptors();
    List<String> suggestableEntities = new ArrayList<String>();

    for (String entity : map.keySet()) {
      suggestableEntities.add(entity);
    }
    return suggestableEntities;
  }

  /**
   * Initialize the AutosuggestionSynchronizer and AggregationSuggestionSynchronizer
   * 
   * @param esConfig
   * @param aaiAdapter
   * @param esAdapter
   * @param nonCachingRestProvider
   */
  private void initAutoSuggestionSynchronizer(ElasticSearchConfig esConfig,
      ActiveInventoryAdapter aaiAdapter, ElasticSearchAdapter esAdapter,
      RestfulDataAccessor nonCachingRestProvider) {
    LOG.info(AaiUiMsgs.INFO_GENERIC, "initAutoSuggestionSynchronizer");

    // Initialize for entityautosuggestindex
    try {
      IndexIntegrityValidator autoSuggestionIndexValidator =
          new IndexIntegrityValidator(nonCachingRestProvider, esConfig.getAutosuggestIndexname(),
              esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
              esConfig.buildAutosuggestionTableConfig());

      syncController.registerIndexValidator(autoSuggestionIndexValidator);

      AutosuggestionSynchronizer suggestionSynchronizer =
          new AutosuggestionSynchronizer(esConfig.getAutosuggestIndexname());
      suggestionSynchronizer.setAaiDataProvider(aaiAdapter);
      suggestionSynchronizer.setEsDataProvider(esAdapter);
      syncController.registerEntitySynchronizer(suggestionSynchronizer);

      AggregationSuggestionSynchronizer aggregationSuggestionSynchronizer =
          new AggregationSuggestionSynchronizer(esConfig.getAutosuggestIndexname());
      aggregationSuggestionSynchronizer.setEsDataProvider(esAdapter);
      syncController.registerEntitySynchronizer(aggregationSuggestionSynchronizer);

      IndexCleaner autosuggestIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
          esConfig.getAutosuggestIndexname(), esConfig.getType(), esConfig.getIpAddress(),
          esConfig.getHttpPort(), syncConfig.getScrollContextTimeToLiveInMinutes(),
          syncConfig.getNumScrollContextItemsToRetrievePerRequest());

      syncController.registerIndexCleaner(autosuggestIndexCleaner);
    } catch (Exception exc) {
      String message = "Error: failed to sync with message = " + exc.getMessage();
      LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
    }
  }

  /**
   * Initialize the AggregationSynchronizer
   * 
   * @param esConfig
   * @param aaiAdapter
   * @param esAdapter
   * @param nonCachingRestProvider
   */
  private void initAggregationSynchronizer(ElasticSearchConfig esConfig,
      ActiveInventoryAdapter aaiAdapter, ElasticSearchAdapter esAdapter,
      RestfulDataAccessor nonCachingRestProvider) {
    LOG.info(AaiUiMsgs.INFO_GENERIC, "initAggregationSynchronizer");

    List<String> aggregationEntities = getAutosuggestableEntitiesFromOXM();

    // For each index: create an IndexValidator, a Synchronizer, and an IndexCleaner
    for (String entity : aggregationEntities) {
      try {
        String indexName = TierSupportUiConstants.getAggregationIndexName(entity);

        IndexIntegrityValidator aggregationIndexValidator = new IndexIntegrityValidator(
            nonCachingRestProvider, indexName, esConfig.getType(), esConfig.getIpAddress(),
            esConfig.getHttpPort(), esConfig.buildAggregationTableConfig());

        syncController.registerIndexValidator(aggregationIndexValidator);

        /*
         * TODO: This per-entity-synchronizer approach will eventually result in AAI / ES overload
         * because of the existing dedicated thread pools for ES + AAI operations within the
         * synchronizer. If we had 50 types to sync then the thread pools within each Synchronizer
         * would cause some heartburn as there would be hundreds of threads trying to talk to AAI.
         * Given that we our running out of time, let's make sure we can get it functional and then
         * we'll re-visit.
         */
        AggregationSynchronizer aggSynchronizer = new AggregationSynchronizer(entity, indexName);
        aggSynchronizer.setAaiDataProvider(aaiAdapter);
        aggSynchronizer.setEsDataProvider(esAdapter);
        syncController.registerEntitySynchronizer(aggSynchronizer);

        IndexCleaner entityDataIndexCleaner = new ElasticSearchIndexCleaner(nonCachingRestProvider,
            indexName, esConfig.getType(), esConfig.getIpAddress(), esConfig.getHttpPort(),
            syncConfig.getScrollContextTimeToLiveInMinutes(),
            syncConfig.getNumScrollContextItemsToRetrievePerRequest());

        syncController.registerIndexCleaner(entityDataIndexCleaner);

      } catch (Exception exc) {
        String message = "Error: failed to sync with message = " + exc.getMessage();
        LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
      }
    }
  }

  /**
   * Instantiates a new sync helper.
   *
   * @param loader the loader
   */
  public SyncHelper(OxmModelLoader loader) {
    try {
      this.contextMap = MDC.getCopyOfContextMap();
      this.syncConfig = SynchronizerConfiguration.getConfig();
      this.esConfig = ElasticSearchConfig.getConfig();
      this.oxmModelLoader = loader;

      UncaughtExceptionHandler uncaughtExceptionHandler = new Thread.UncaughtExceptionHandler() {

        @Override
        public void uncaughtException(Thread thread, Throwable exc) {
          LOG.error(AaiUiMsgs.ERROR_GENERIC, thread.getName() + ": " + exc);
        }
      };

      ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("SyncHelper-%d")
          .setUncaughtExceptionHandler(uncaughtExceptionHandler).build();

      periodicExecutor = Executors.newScheduledThreadPool(3, namedThreadFactory);

      /*
       * We only want to initialize the synchronizer if sync has been configured to start
       */
      if (syncConfig.isConfigOkForStartupSync() || syncConfig.isConfigOkForPeriodicSync()) {
        initializeSyncController();
      }

      // schedule startup synchronization
      if (syncConfig.isConfigOkForStartupSync()) {

        long taskInitialDelayInMs = syncConfig.getSyncTaskInitialDelayInMs();
        if (taskInitialDelayInMs != SynchronizerConstants.DELAY_NO_STARTUP_SYNC_IN_MS) {
          oneShotExecutor.schedule(new SyncTask(true), taskInitialDelayInMs, TimeUnit.MILLISECONDS);
          LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine startup synchronization is enabled.");
        } else {
          LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine startup synchronization is disabled.");
        }
      }

      // schedule periodic synchronization
      if (syncConfig.isConfigOkForPeriodicSync()) {

        TimeZone tz = TimeZone.getTimeZone(syncConfig.getSyncTaskStartTimeTimeZone());
        Calendar calendar = Calendar.getInstance(tz);
        sdf.setTimeZone(tz);

        calendar.set(Calendar.HOUR_OF_DAY, syncConfig.getSyncTaskStartTimeHr());
        calendar.set(Calendar.MINUTE, syncConfig.getSyncTaskStartTimeMin());
        calendar.set(Calendar.SECOND, syncConfig.getSyncTaskStartTimeSec());

        long timeCurrent = calendar.getTimeInMillis();
        int taskFrequencyInDay = syncConfig.getSyncTaskFrequencyInDay();
        timeNextSync.getAndSet(getFirstSyncTime(calendar, timeCurrent, taskFrequencyInDay));

        long delayUntilFirstRegSyncInMs = 0;
        delayUntilFirstRegSyncInMs = timeNextSync.get() - timeCurrent;

        // Do all calculation in milliseconds
        long taskFreqencyInMs = taskFrequencyInDay * SynchronizerConstants.MILLISEC_IN_A_DAY;

        if (taskFreqencyInMs != SynchronizerConstants.DELAY_NO_PERIODIC_SYNC_IN_MS) {
          periodicExecutor.scheduleAtFixedRate(new SyncTask(false), delayUntilFirstRegSyncInMs,
              taskFreqencyInMs, TimeUnit.MILLISECONDS);
          LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine periodic synchronization is enabled.");
          // case: when - startup sync is misconfigured or is disabled
          // - give a clue to user when is the next periodic sync
          if (!syncConfig.isConfigOkForStartupSync()
              || syncConfig.isConfigDisabledForInitialSync()) {
            LOG.info(AaiUiMsgs.SYNC_TO_BEGIN, syncController.getControllerName(),
                sdf.format(timeNextSync).replaceAll(SynchronizerConstants.TIME_STD,
                    SynchronizerConstants.TIME_CONFIG_STD));
          }
        } else {
          LOG.info(AaiUiMsgs.INFO_GENERIC, "Search Engine periodic synchronization is disabled.");
        }
      }

    } catch (Exception exc) {
      String message = "Caught an exception while starting up the SyncHelper. Error cause = \n"
          + ErrorUtil.extractStackTraceElements(5, exc);
      LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
    }
  }


  /**
   * Shutdown.
   */
  public void shutdown() {

    if (oneShotExecutor != null) {
      oneShotExecutor.shutdown();
    }

    if (periodicExecutor != null) {
      periodicExecutor.shutdown();
    }

    if (historicalExecutor != null) {
      historicalExecutor.shutdown();
    }

    if (syncController != null) {
      syncController.shutdown();
    }

    if (entityCounterHistorySummarizer != null) {
      entityCounterHistorySummarizer.shutdown();
    }

  }

  public OxmModelLoader getOxmModelLoader() {
    return oxmModelLoader;
  }

  public void setOxmModelLoader(OxmModelLoader oxmModelLoader) {
    this.oxmModelLoader = oxmModelLoader;
  }
}