diff options
Diffstat (limited to 'sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync')
4 files changed, 1501 insertions, 0 deletions
diff --git a/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSyncControllerFactory.java b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSyncControllerFactory.java new file mode 100644 index 0000000..8681853 --- /dev/null +++ b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSyncControllerFactory.java @@ -0,0 +1,241 @@ +/** + * ============LICENSE_START=================================================== + * SPARKY (AAI UI service) + * ============================================================================ + * Copyright © 2017 AT&T Intellectual Property. + * Copyright © 2017 Amdocs + * All rights reserved. + * ============================================================================ + * 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 and OpenECOMP are trademarks + * and service marks of AT&T Intellectual Property. + */ +package org.onap.aai.sparky.aggregation.sync; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.onap.aai.cl.api.Logger; +import org.onap.aai.cl.eelf.LoggerFactory; +import org.onap.aai.sparky.config.oxm.OxmEntityLookup; +import org.onap.aai.sparky.config.oxm.SuggestionEntityDescriptor; +import org.onap.aai.sparky.config.oxm.SuggestionEntityLookup; +import org.onap.aai.sparky.dal.ActiveInventoryAdapter; +import org.onap.aai.sparky.dal.ElasticSearchAdapter; +import org.onap.aai.sparky.logging.AaiUiMsgs; +import org.onap.aai.sparky.sync.ElasticSearchIndexCleaner; +import org.onap.aai.sparky.sync.ElasticSearchSchemaFactory; +import org.onap.aai.sparky.sync.IndexCleaner; +import org.onap.aai.sparky.sync.IndexIntegrityValidator; +import org.onap.aai.sparky.sync.SyncController; +import org.onap.aai.sparky.sync.SyncControllerImpl; +import org.onap.aai.sparky.sync.SyncControllerRegistrar; +import org.onap.aai.sparky.sync.SyncControllerRegistry; +import org.onap.aai.sparky.sync.config.ElasticSearchEndpointConfig; +import org.onap.aai.sparky.sync.config.ElasticSearchSchemaConfig; +import org.onap.aai.sparky.sync.config.NetworkStatisticsConfig; +import org.onap.aai.sparky.sync.config.SyncControllerConfig; + +public class AggregationSyncControllerFactory implements SyncControllerRegistrar { + + private static final Logger LOG = + LoggerFactory.getInstance().getLogger(AggregationSyncControllerFactory.class); + + private ActiveInventoryAdapter aaiAdapter; + private ElasticSearchAdapter esAdapter; + private SuggestionEntityLookup suggestionEntityLookup; + + private Map<String, String> aggregationEntityToIndexMap; + private Map<String, ElasticSearchSchemaConfig> indexNameToSchemaConfigMap; + + private ElasticSearchEndpointConfig elasticSearchEndpointConfig; + private SyncControllerConfig syncControllerConfig; + private SyncControllerRegistry syncControllerRegistry; + private NetworkStatisticsConfig aaiStatConfig; + private NetworkStatisticsConfig esStatConfig; + private OxmEntityLookup oxmEntityLookup; + private ElasticSearchSchemaFactory elasticSearchSchemaFactory; + + private List<SyncController> syncControllers; + + public AggregationSyncControllerFactory(ElasticSearchEndpointConfig esEndpointConfig, + SyncControllerConfig syncControllerConfig, SyncControllerRegistry syncControllerRegistry, + SuggestionEntityLookup suggestionEntityLookup, + OxmEntityLookup oxmEntityLookup, + ElasticSearchSchemaFactory elasticSearchSchemaFactory) { + this.elasticSearchSchemaFactory = elasticSearchSchemaFactory; + this.syncControllers = new ArrayList<SyncController>(); + this.elasticSearchEndpointConfig = esEndpointConfig; + this.syncControllerConfig = syncControllerConfig; + this.syncControllerRegistry = syncControllerRegistry; + this.suggestionEntityLookup = suggestionEntityLookup; + this.oxmEntityLookup = oxmEntityLookup; + } + + public NetworkStatisticsConfig getAaiStatConfig() { + return aaiStatConfig; + } + + public void setAaiStatConfig(NetworkStatisticsConfig aaiStatConfig) { + this.aaiStatConfig = aaiStatConfig; + } + + public NetworkStatisticsConfig getEsStatConfig() { + return esStatConfig; + } + + public void setEsStatConfig(NetworkStatisticsConfig esStatConfig) { + this.esStatConfig = esStatConfig; + } + + public Map<String, ElasticSearchSchemaConfig> getIndexNameToSchemaConfigMap() { + return indexNameToSchemaConfigMap; + } + + public void setIndexNameToSchemaConfigMap( + Map<String, ElasticSearchSchemaConfig> indexNameToSchemaConfigMap) { + this.indexNameToSchemaConfigMap = indexNameToSchemaConfigMap; + } + + public ElasticSearchEndpointConfig getElasticSearchEndpointConfig() { + return elasticSearchEndpointConfig; + } + + public void setElasticSearchEndpointConfig( + ElasticSearchEndpointConfig elasticSearchEndpointConfig) { + this.elasticSearchEndpointConfig = elasticSearchEndpointConfig; + } + + public SyncControllerConfig getSyncControllerConfig() { + return syncControllerConfig; + } + + public void setSyncControllerConfig(SyncControllerConfig syncControllerConfig) { + this.syncControllerConfig = syncControllerConfig; + } + + public ActiveInventoryAdapter getAaiAdapter() { + return aaiAdapter; + } + + public void setAaiAdapter(ActiveInventoryAdapter aaiAdapter) { + this.aaiAdapter = aaiAdapter; + } + + public ElasticSearchAdapter getEsAdapter() { + return esAdapter; + } + + public void setEsAdapter(ElasticSearchAdapter esAdapter) { + this.esAdapter = esAdapter; + } + + public SuggestionEntityLookup getSuggestionEntityLookup() { + return suggestionEntityLookup; + } + + public void setSuggestionEntityLookup(SuggestionEntityLookup suggestionEntityLookup) { + this.suggestionEntityLookup = suggestionEntityLookup; + } + + public Map<String, String> getAggregationEntityToIndexMap() { + return aggregationEntityToIndexMap; + } + + public void setAggregationEntityToIndexMap(Map<String, String> aggregationEntityToIndexMap) { + this.aggregationEntityToIndexMap = aggregationEntityToIndexMap; + } + + public void buildControllers() { + + if (syncControllerConfig.isEnabled()) { + + Map<String, SuggestionEntityDescriptor> suggestionEntitites = + suggestionEntityLookup.getSuggestionSearchEntityDescriptors(); + SyncControllerImpl aggregationSyncController = null; + + for (String entityType : suggestionEntitites.keySet()) { + + String indexName = aggregationEntityToIndexMap.get(entityType); + + if (indexName == null) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "Could not determine aggregation index name" + " for entity type: " + entityType); + continue; + } + + try { + + aggregationSyncController = new SyncControllerImpl(syncControllerConfig, entityType); + + ElasticSearchSchemaConfig schemaConfig = indexNameToSchemaConfigMap.get(indexName); + + if (schemaConfig == null) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "Could not determine elastic search schema config for index name: " + indexName); + continue; + } + + IndexIntegrityValidator aggregationIndexValidator = new IndexIntegrityValidator(esAdapter, + schemaConfig, elasticSearchEndpointConfig, elasticSearchSchemaFactory.getIndexSchema(schemaConfig)); + + aggregationSyncController.registerIndexValidator(aggregationIndexValidator); + + AggregationSynchronizer aggSynchronizer = new AggregationSynchronizer(entityType, + schemaConfig, syncControllerConfig.getNumInternalSyncWorkers(), + syncControllerConfig.getNumSyncActiveInventoryWorkers(), + syncControllerConfig.getNumSyncElasticWorkers(), aaiStatConfig, esStatConfig, + oxmEntityLookup); + + aggSynchronizer.setAaiAdapter(aaiAdapter); + aggSynchronizer.setElasticSearchAdapter(esAdapter); + + aggregationSyncController.registerEntitySynchronizer(aggSynchronizer); + + IndexCleaner entityDataIndexCleaner = + new ElasticSearchIndexCleaner(esAdapter, elasticSearchEndpointConfig, schemaConfig); + + aggregationSyncController.registerIndexCleaner(entityDataIndexCleaner); + + syncControllers.add(aggregationSyncController); + } catch (Exception exc) { + + exc.printStackTrace(); + + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "Failed to build aggregation sync controller. Error : " + exc.getMessage()); + } + + } + } else { + LOG.info(AaiUiMsgs.INFO_GENERIC, "Sync controller with name = " + + syncControllerConfig.getControllerName() + " is disabled"); + } + } + + @Override + public void registerController() { + + buildControllers(); + + if ( syncControllerRegistry != null ) { + for ( SyncController controller : syncControllers ) { + syncControllerRegistry.registerSyncController(controller); + } + } + + } +} diff --git a/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSynchronizer.java b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSynchronizer.java new file mode 100644 index 0000000..a438215 --- /dev/null +++ b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/AggregationSynchronizer.java @@ -0,0 +1,782 @@ +/** + * ============LICENSE_START=================================================== + * SPARKY (AAI UI service) + * ============================================================================ + * Copyright © 2017 AT&T Intellectual Property. + * Copyright © 2017 Amdocs + * All rights reserved. + * ============================================================================ + * 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 and OpenECOMP are trademarks + * and service marks of AT&T Intellectual Property. + */ +package org.onap.aai.sparky.aggregation.sync; + +import static java.util.concurrent.CompletableFuture.supplyAsync; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Deque; +import java.util.EnumSet; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import org.onap.aai.cl.api.Logger; +import org.onap.aai.cl.eelf.LoggerFactory; +import org.onap.aai.cl.mdc.MdcContext; +import org.onap.aai.restclient.client.OperationResult; +import org.onap.aai.sparky.config.oxm.OxmEntityDescriptor; +import org.onap.aai.sparky.config.oxm.OxmEntityLookup; +import org.onap.aai.sparky.dal.ActiveInventoryAdapter; +import org.onap.aai.sparky.dal.NetworkTransaction; +import org.onap.aai.sparky.dal.rest.HttpMethod; +import org.onap.aai.sparky.logging.AaiUiMsgs; +import org.onap.aai.sparky.sync.AbstractEntitySynchronizer; +import org.onap.aai.sparky.sync.IndexSynchronizer; +import org.onap.aai.sparky.sync.SynchronizerConstants; +import org.onap.aai.sparky.sync.config.ElasticSearchSchemaConfig; +import org.onap.aai.sparky.sync.config.NetworkStatisticsConfig; +import org.onap.aai.sparky.sync.entity.AggregationEntity; +import org.onap.aai.sparky.sync.entity.MergableEntity; +import org.onap.aai.sparky.sync.entity.SelfLinkDescriptor; +import org.onap.aai.sparky.sync.enumeration.OperationState; +import org.onap.aai.sparky.sync.enumeration.SynchronizerState; +import org.onap.aai.sparky.sync.task.PerformActiveInventoryRetrieval; +import org.onap.aai.sparky.sync.task.PerformElasticSearchPut; +import org.onap.aai.sparky.sync.task.PerformElasticSearchRetrieval; +import org.onap.aai.sparky.sync.task.PerformElasticSearchUpdate; +import org.onap.aai.sparky.util.NodeUtils; +import org.slf4j.MDC; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectReader; +import com.fasterxml.jackson.databind.node.ArrayNode; + +/** + * The Class AutosuggestionSynchronizer. + */ +public class AggregationSynchronizer extends AbstractEntitySynchronizer + implements IndexSynchronizer { + + /** + * The Class RetryAggregationEntitySyncContainer. + */ + private class RetryAggregationEntitySyncContainer { + NetworkTransaction txn; + AggregationEntity ae; + + /** + * Instantiates a new retry aggregation entity sync container. + * + * @param txn the txn + * @param ae the se + */ + public RetryAggregationEntitySyncContainer(NetworkTransaction txn, AggregationEntity ae) { + this.txn = txn; + this.ae = ae; + } + + public NetworkTransaction getNetworkTransaction() { + return txn; + } + + public AggregationEntity getAggregationEntity() { + return ae; + } + } + + private static final Logger LOG = + LoggerFactory.getInstance().getLogger(AggregationSynchronizer.class); + private static final String INSERTION_DATE_TIME_FORMAT = "yyyyMMdd'T'HHmmssZ"; + + private boolean allWorkEnumerated; + private Deque<SelfLinkDescriptor> selflinks; + private Deque<RetryAggregationEntitySyncContainer> retryQueue; + private Map<String, Integer> retryLimitTracker; + protected ExecutorService esPutExecutor; + private ConcurrentHashMap<String, AtomicInteger> entityCounters; + private boolean syncInProgress; + private Map<String, String> contextMap; + private String entityType; + private ElasticSearchSchemaConfig schemaConfig; + private OxmEntityLookup oxmEntityLookup; + + /** + * Instantiates a new entity aggregation synchronizer. + * + * @param indexName the index name + * @throws Exception the exception + */ + public AggregationSynchronizer(String entityType, ElasticSearchSchemaConfig schemaConfig, + int numSyncWorkers, int numActiveInventoryWorkers, int numElasticWorkers, + NetworkStatisticsConfig aaiStatConfig, NetworkStatisticsConfig esStatConfig, + OxmEntityLookup oxmEntityLookup) throws Exception { + + super(LOG, "AGGES-" + schemaConfig.getIndexName().toUpperCase(), numSyncWorkers, + numActiveInventoryWorkers, numElasticWorkers, schemaConfig.getIndexName(),aaiStatConfig, esStatConfig); + + this.oxmEntityLookup = oxmEntityLookup; + + this.schemaConfig = schemaConfig; + this.entityType = entityType; + this.allWorkEnumerated = false; + this.entityCounters = new ConcurrentHashMap<String, AtomicInteger>(); + this.synchronizerName = "Entity Aggregation Synchronizer"; + this.enabledStatFlags = EnumSet.of(StatFlag.AAI_REST_STATS, StatFlag.ES_REST_STATS); + this.syncInProgress = false; + this.allWorkEnumerated = false; + this.selflinks = new ConcurrentLinkedDeque<SelfLinkDescriptor>(); + this.retryQueue = new ConcurrentLinkedDeque<RetryAggregationEntitySyncContainer>(); + this.retryLimitTracker = new ConcurrentHashMap<String, Integer>(); + + this.esPutExecutor = NodeUtils.createNamedExecutor("AGGES-ES-PUT", 1, LOG); + + this.aaiEntityStats.intializeEntityCounters(entityType); + this.esEntityStats.intializeEntityCounters(entityType); + + this.contextMap = MDC.getCopyOfContextMap(); + } + + /** + * Collect all the work. + * + * @return the operation state + */ + private OperationState collectAllTheWork() { + final Map<String, String> contextMap = MDC.getCopyOfContextMap(); + final String entity = this.getEntityType(); + try { + + aaiWorkOnHand.set(1); + + supplyAsync(new Supplier<Void>() { + + @Override + public Void get() { + MDC.setContextMap(contextMap); + OperationResult typeLinksResult = null; + try { + typeLinksResult = aaiAdapter.getSelfLinksByEntityType(entity); + aaiWorkOnHand.decrementAndGet(); + processEntityTypeSelfLinks(typeLinksResult); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, "Processing execption while building working set. Error:" + + exc.getMessage()); + } + + return null; + } + + }, aaiExecutor).whenComplete((result, error) -> { + + if (error != null) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "An error occurred getting data from AAI. Error = " + error.getMessage()); + } + }); + + while (aaiWorkOnHand.get() != 0) { + + if (LOG.isDebugEnabled()) { + LOG.debug(AaiUiMsgs.WAIT_FOR_ALL_SELFLINKS_TO_BE_COLLECTED); + } + + Thread.sleep(1000); + } + + aaiWorkOnHand.set(selflinks.size()); + allWorkEnumerated = true; + syncEntityTypes(); + + while (!isSyncDone()) { + performRetrySync(); + Thread.sleep(1000); + } + + /* + * Make sure we don't hang on to retries that failed which could cause issues during future + * syncs + */ + retryLimitTracker.clear(); + + } catch (Exception exc) { + // TODO -> LOG, waht should be logged here? + } + + return OperationState.OK; + } + + + /** + * Perform retry sync. + */ + private void performRetrySync() { + while (retryQueue.peek() != null) { + + RetryAggregationEntitySyncContainer rsc = retryQueue.poll(); + if (rsc != null) { + + AggregationEntity ae = rsc.getAggregationEntity(); + NetworkTransaction txn = rsc.getNetworkTransaction(); + + String link = null; + try { + /* + * In this retry flow the se object has already derived its fields + */ + link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), ae.getId()); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_URI, exc.getLocalizedMessage()); + } + + if (link != null) { + NetworkTransaction retryTransaction = new NetworkTransaction(); + retryTransaction.setLink(link); + retryTransaction.setEntityType(txn.getEntityType()); + retryTransaction.setDescriptor(txn.getDescriptor()); + retryTransaction.setOperationType(HttpMethod.GET); + + /* + * IMPORTANT - DO NOT incrementAndGet the esWorkOnHand as this is a retry flow! We already + * called incrementAndGet when queuing the failed PUT! + */ + + supplyAsync(new PerformElasticSearchRetrieval(retryTransaction, elasticSearchAdapter), + esExecutor).whenComplete((result, error) -> { + + esWorkOnHand.decrementAndGet(); + + if (error != null) { + LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED_RESYNC, error.getLocalizedMessage()); + } else { + updateElasticSearchCounters(result); + performDocumentUpsert(result, ae); + } + }); + } + + } + } + } + + /** + * Perform document upsert. + * + * @param esGetTxn the es get txn + * @param ae the ae + */ + protected void performDocumentUpsert(NetworkTransaction esGetTxn, AggregationEntity ae) { + /** + * <p> + * <ul> + * As part of the response processing we need to do the following: + * <li>1. Extract the version (if present), it will be the ETAG when we use the + * Search-Abstraction-Service + * <li>2. Spawn next task which is to do the PUT operation into elastic with or with the version + * tag + * <li>a) if version is null or RC=404, then standard put, no _update with version tag + * <li>b) if version != null, do PUT with _update?version= versionNumber in the URI to elastic + * </ul> + * </p> + */ + String link = null; + try { + link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), ae.getId()); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ES_LINK_UPSERT, exc.getLocalizedMessage()); + return; + } + + String versionNumber = null; + boolean wasEntryDiscovered = false; + if (esGetTxn.getOperationResult().getResultCode() == 404) { + LOG.info(AaiUiMsgs.ES_SIMPLE_PUT, ae.getEntityPrimaryKeyValue()); + } else if (esGetTxn.getOperationResult().getResultCode() == 200) { + wasEntryDiscovered = true; + try { + versionNumber = NodeUtils.extractFieldValueFromObject( + NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()), + "_version"); + } catch (IOException exc) { + String message = + "Error extracting version number from response, aborting aggregation entity sync of " + + ae.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage(); + LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message); + return; + } + } else { + /* + * Not being a 200 does not mean a failure. eg 201 is returned for created. TODO -> Should we + * return. + */ + LOG.error(AaiUiMsgs.ES_OPERATION_RETURN_CODE, + String.valueOf(esGetTxn.getOperationResult().getResultCode())); + return; + } + + try { + String jsonPayload = null; + if (wasEntryDiscovered) { + try { + ArrayList<JsonNode> sourceObject = new ArrayList<JsonNode>(); + NodeUtils.extractObjectsByKey( + NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()), + "_source", sourceObject); + + if (!sourceObject.isEmpty()) { + String responseSource = NodeUtils.convertObjectToJson(sourceObject.get(0), false); + MergableEntity me = mapper.readValue(responseSource, MergableEntity.class); + ObjectReader updater = mapper.readerForUpdating(me); + MergableEntity merged = updater.readValue(ae.getAsJson()); + jsonPayload = mapper.writeValueAsString(merged); + } + } catch (IOException exc) { + String message = + "Error extracting source value from response, aborting aggregation entity sync of " + + ae.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage(); + LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message); + return; + } + } else { + jsonPayload = ae.getAsJson(); + } + + if (wasEntryDiscovered) { + if (versionNumber != null && jsonPayload != null) { + + String requestPayload = + elasticSearchAdapter.buildBulkImportOperationRequest(schemaConfig.getIndexName(), + schemaConfig.getIndexDocType(), ae.getId(), versionNumber, jsonPayload); + + NetworkTransaction transactionTracker = new NetworkTransaction(); + transactionTracker.setEntityType(esGetTxn.getEntityType()); + transactionTracker.setDescriptor(esGetTxn.getDescriptor()); + transactionTracker.setOperationType(HttpMethod.PUT); + + esWorkOnHand.incrementAndGet(); + supplyAsync(new PerformElasticSearchUpdate(elasticSearchAdapter.getBulkUrl(), + requestPayload, elasticSearchAdapter, transactionTracker), esPutExecutor) + .whenComplete((result, error) -> { + + esWorkOnHand.decrementAndGet(); + + if (error != null) { + String message = "Aggregation entity sync UPDATE PUT error - " + + error.getLocalizedMessage(); + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + } else { + updateElasticSearchCounters(result); + processStoreDocumentResult(result, esGetTxn, ae); + } + }); + } + + } else { + if (link != null && jsonPayload != null) { + + NetworkTransaction updateElasticTxn = new NetworkTransaction(); + updateElasticTxn.setLink(link); + updateElasticTxn.setEntityType(esGetTxn.getEntityType()); + updateElasticTxn.setDescriptor(esGetTxn.getDescriptor()); + updateElasticTxn.setOperationType(HttpMethod.PUT); + + esWorkOnHand.incrementAndGet(); + supplyAsync(new PerformElasticSearchPut(jsonPayload, updateElasticTxn, elasticSearchAdapter), + esPutExecutor).whenComplete((result, error) -> { + + esWorkOnHand.decrementAndGet(); + + if (error != null) { + String message = + "Aggregation entity sync UPDATE PUT error - " + error.getLocalizedMessage(); + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + } else { + updateElasticSearchCounters(result); + processStoreDocumentResult(result, esGetTxn, ae); + } + }); + } + } + } catch (Exception exc) { + String message = "Exception caught during aggregation entity sync PUT operation. Message - " + + exc.getLocalizedMessage(); + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + } + } + + /** + * Should allow retry. + * + * @param id the id + * @return true, if successful + */ + private boolean shouldAllowRetry(String id) { + boolean isRetryAllowed = true; + if (retryLimitTracker.get(id) != null) { + Integer currentCount = retryLimitTracker.get(id); + if (currentCount.intValue() >= RETRY_COUNT_PER_ENTITY_LIMIT.intValue()) { + isRetryAllowed = false; + String message = "Aggregation entity re-sync limit reached for " + id + + ", re-sync will no longer be attempted for this entity"; + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + } else { + Integer newCount = new Integer(currentCount.intValue() + 1); + retryLimitTracker.put(id, newCount); + } + } else { + Integer firstRetryCount = new Integer(1); + retryLimitTracker.put(id, firstRetryCount); + } + + return isRetryAllowed; + } + + /** + * Process store document result. + * + * @param esPutResult the es put result + * @param esGetResult the es get result + * @param ae the ae + */ + private void processStoreDocumentResult(NetworkTransaction esPutResult, + NetworkTransaction esGetResult, AggregationEntity ae) { + + OperationResult or = esPutResult.getOperationResult(); + + if (!or.wasSuccessful()) { + if (or.getResultCode() == VERSION_CONFLICT_EXCEPTION_CODE) { + + if (shouldAllowRetry(ae.getId())) { + esWorkOnHand.incrementAndGet(); + + RetryAggregationEntitySyncContainer rsc = + new RetryAggregationEntitySyncContainer(esGetResult, ae); + retryQueue.push(rsc); + + String message = "Store document failed during aggregation entity synchronization" + + " due to version conflict. Entity will be re-synced."; + LOG.warn(AaiUiMsgs.ERROR_GENERIC, message); + } + } else { + String message = + "Store document failed during aggregation entity synchronization with result code " + + or.getResultCode() + " and result message " + or.getResult(); + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + } + } + } + + /** + * Sync entity types. + */ + private void syncEntityTypes() { + + while (selflinks.peek() != null) { + + SelfLinkDescriptor linkDescriptor = selflinks.poll(); + aaiWorkOnHand.decrementAndGet(); + + OxmEntityDescriptor descriptor = null; + + if (linkDescriptor.getSelfLink() != null && linkDescriptor.getEntityType() != null) { + + descriptor = oxmEntityLookup.getEntityDescriptors().get(linkDescriptor.getEntityType()); + + if (descriptor == null) { + LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, linkDescriptor.getEntityType()); + // go to next element in iterator + continue; + } + + NetworkTransaction txn = new NetworkTransaction(); + txn.setDescriptor(descriptor); + txn.setLink(linkDescriptor.getSelfLink()); + txn.setOperationType(HttpMethod.GET); + txn.setEntityType(linkDescriptor.getEntityType()); + + aaiWorkOnHand.incrementAndGet(); + + supplyAsync(new PerformActiveInventoryRetrieval(txn, aaiAdapter), aaiExecutor) + .whenComplete((result, error) -> { + + aaiWorkOnHand.decrementAndGet(); + + if (error != null) { + LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_GENERIC, error.getLocalizedMessage()); + } else { + if (result == null) { + LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_FOR_SELF_LINK, + linkDescriptor.getSelfLink()); + } else { + updateActiveInventoryCounters(result); + fetchDocumentForUpsert(result); + } + } + }); + } + + } + + } + + /** + * Fetch document for upsert. + * + * @param txn the txn + */ + private void fetchDocumentForUpsert(NetworkTransaction txn) { + // modified + if (!txn.getOperationResult().wasSuccessful()) { + String message = "Self link failure. Result - " + txn.getOperationResult().getResult(); + LOG.error(AaiUiMsgs.ERROR_GENERIC, message); + return; + } + + try { + final String jsonResult = txn.getOperationResult().getResult(); + if (jsonResult != null && jsonResult.length() > 0) { + + AggregationEntity ae = new AggregationEntity(); + ae.setLink(ActiveInventoryAdapter.extractResourcePath(txn.getLink())); + populateAggregationEntityDocument(ae, jsonResult, txn.getDescriptor()); + ae.deriveFields(); + + String link = null; + try { + link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), ae.getId()); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_QUERY, exc.getLocalizedMessage()); + } + + if (link != null) { + NetworkTransaction n2 = new NetworkTransaction(); + n2.setLink(link); + n2.setEntityType(txn.getEntityType()); + n2.setDescriptor(txn.getDescriptor()); + n2.setOperationType(HttpMethod.GET); + + esWorkOnHand.incrementAndGet(); + + supplyAsync(new PerformElasticSearchRetrieval(n2, elasticSearchAdapter), esExecutor) + .whenComplete((result, error) -> { + + esWorkOnHand.decrementAndGet(); + + if (error != null) { + LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED, error.getLocalizedMessage()); + } else { + updateElasticSearchCounters(result); + performDocumentUpsert(result, ae); + } + }); + } + } + + } catch (JsonProcessingException exc) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "There was a JSON processing error fetching the elastic document for upsert. Error: " + + exc.getMessage()); + } catch (IOException exc) { + LOG.error(AaiUiMsgs.ERROR_GENERIC, + "There was an IO error fetching the elastic document for upsert. Error: " + exc.getMessage()); + } + } + + + /** + * Populate aggregation entity document. + * + * @param doc the doc + * @param result the result + * @param resultDescriptor the result descriptor + * @throws JsonProcessingException the json processing exception + * @throws IOException Signals that an I/O exception has occurred. + */ + protected void populateAggregationEntityDocument(AggregationEntity doc, String result, + OxmEntityDescriptor resultDescriptor) throws JsonProcessingException, IOException { + doc.setEntityType(resultDescriptor.getEntityName()); + JsonNode entityNode = mapper.readTree(result); + Map<String, Object> map = mapper.convertValue(entityNode, Map.class); + doc.copyAttributeKeyValuePair(map); + } + + /** + * Process entity type self links. + * + * @param operationResult the operation result + */ + private void processEntityTypeSelfLinks(OperationResult operationResult) { + + JsonNode rootNode = null; + + if ( operationResult == null ) { + return; + } + + final String jsonResult = operationResult.getResult(); + + if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) { + + try { + rootNode = mapper.readTree(jsonResult); + } catch (IOException exc) { + String message = + "Could not deserialize JSON (representing operation result) as node tree. " + + "Operation result = " + jsonResult + ". " + exc.getLocalizedMessage(); + LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, message); + } + + JsonNode resultData = rootNode.get("result-data"); + ArrayNode resultDataArrayNode = null; + + if (resultData.isArray()) { + resultDataArrayNode = (ArrayNode) resultData; + + Iterator<JsonNode> elementIterator = resultDataArrayNode.elements(); + JsonNode element = null; + + while (elementIterator.hasNext()) { + element = elementIterator.next(); + + final String resourceType = NodeUtils.getNodeFieldAsText(element, "resource-type"); + final String resourceLink = NodeUtils.getNodeFieldAsText(element, "resource-link"); + + OxmEntityDescriptor descriptor = null; + + if (resourceType != null && resourceLink != null) { + + descriptor = oxmEntityLookup.getEntityDescriptors().get(resourceType); + + if (descriptor == null) { + LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, resourceType); + // go to next element in iterator + continue; + } + + selflinks.add(new SelfLinkDescriptor(resourceLink, SynchronizerConstants.NODES_ONLY_MODIFIER, resourceType)); + + + } + } + } + } + + } + + /* + * (non-Javadoc) + * + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync() + */ + @Override + public OperationState doSync() { + this.syncDurationInMs = -1; + syncStartedTimeStampInMs = System.currentTimeMillis(); + String txnID = NodeUtils.getRandomTxnId(); + MdcContext.initialize(txnID, "AggregationSynchronizer", "", "Sync", ""); + + return collectAllTheWork(); + } + + @Override + public SynchronizerState getState() { + + if (!isSyncDone()) { + return SynchronizerState.PERFORMING_SYNCHRONIZATION; + } + + return SynchronizerState.IDLE; + + } + + /* + * (non-Javadoc) + * + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean) + */ + @Override + public String getStatReport(boolean showFinalReport) { + syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs; + return getStatReport(syncDurationInMs, showFinalReport); + } + + public String getEntityType() { + return entityType; + } + + public void setEntityType(String entityType) { + this.entityType = entityType; + } + + /* + * (non-Javadoc) + * + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown() + */ + @Override + public void shutdown() { + this.shutdownExecutors(); + } + + @Override + protected boolean isSyncDone() { + + int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get(); + + if (LOG.isDebugEnabled()) { + LOG.debug(AaiUiMsgs.DEBUG_GENERIC, indexName + ", isSyncDone(), totalWorkOnHand = " + + totalWorkOnHand + " all work enumerated = " + allWorkEnumerated); + } + + if (totalWorkOnHand > 0 || !allWorkEnumerated) { + return false; + } + + this.syncInProgress = false; + + return true; + } + + /* + * (non-Javadoc) + * + * @see org.openecomp.sparky.synchronizer.AbstractEntitySynchronizer#clearCache() + */ + @Override + public void clearCache() { + + if (syncInProgress) { + LOG.debug(AaiUiMsgs.DEBUG_GENERIC, + "Autosuggestion Entity Summarizer in progress, request to clear cache ignored"); + return; + } + + super.clearCache(); + this.resetCounters(); + if (entityCounters != null) { + entityCounters.clear(); + } + + allWorkEnumerated = false; + + } + +} diff --git a/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySummarizer.java b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySummarizer.java new file mode 100644 index 0000000..9063e92 --- /dev/null +++ b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySummarizer.java @@ -0,0 +1,384 @@ +/** + * ============LICENSE_START=================================================== + * SPARKY (AAI UI service) + * ============================================================================ + * Copyright © 2017 AT&T Intellectual Property. + * Copyright © 2017 Amdocs + * All rights reserved. + * ============================================================================ + * 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 and OpenECOMP are trademarks + * and service marks of AT&T Intellectual Property. + */ +package org.onap.aai.sparky.aggregation.sync; + +import static java.util.concurrent.CompletableFuture.supplyAsync; + +import java.io.IOException; +import java.sql.Timestamp; +import java.text.SimpleDateFormat; +import java.util.Collection; +import java.util.EnumSet; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import javax.json.Json; +import javax.ws.rs.core.MediaType; + +import org.onap.aai.cl.api.Logger; +import org.onap.aai.cl.eelf.LoggerFactory; +import org.onap.aai.cl.mdc.MdcContext; +import org.onap.aai.restclient.client.OperationResult; +import org.onap.aai.sparky.config.oxm.SearchableEntityLookup; +import org.onap.aai.sparky.config.oxm.SearchableOxmEntityDescriptor; +import org.onap.aai.sparky.dal.rest.HttpMethod; +import org.onap.aai.sparky.logging.AaiUiMsgs; +import org.onap.aai.sparky.sync.AbstractEntitySynchronizer; +import org.onap.aai.sparky.sync.IndexSynchronizer; +import org.onap.aai.sparky.sync.config.ElasticSearchSchemaConfig; +import org.onap.aai.sparky.sync.config.NetworkStatisticsConfig; +import org.onap.aai.sparky.sync.enumeration.OperationState; +import org.onap.aai.sparky.sync.enumeration.SynchronizerState; +import org.onap.aai.sparky.util.NodeUtils; +import org.slf4j.MDC; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; + +/** + * The Class HistoricalEntitySummarizer. + */ +public class HistoricalEntitySummarizer extends AbstractEntitySynchronizer + implements IndexSynchronizer { + + private static final Logger LOG = LoggerFactory.getInstance().getLogger(HistoricalEntitySummarizer.class); + private static final String INSERTION_DATE_TIME_FORMAT = "yyyyMMdd'T'HHmmssZ"; + + private boolean allWorkEnumerated; + private ConcurrentHashMap<String, AtomicInteger> entityCounters; + private boolean syncInProgress; + private Map<String, String> contextMap; + private ElasticSearchSchemaConfig schemaConfig; + private SearchableEntityLookup searchableEntityLookup; + + /** + * Instantiates a new historical entity summarizer. + * + * @param indexName the index name + * @throws Exception the exception + */ + public HistoricalEntitySummarizer(ElasticSearchSchemaConfig schemaConfig, int internalSyncWorkers, + int aaiWorkers, int esWorkers, NetworkStatisticsConfig aaiStatConfig, + NetworkStatisticsConfig esStatConfig, SearchableEntityLookup searchableEntityLookup) + throws Exception { + super(LOG, "HES", internalSyncWorkers, aaiWorkers, esWorkers, schemaConfig.getIndexName(), aaiStatConfig, esStatConfig); + + this.schemaConfig = schemaConfig; + this.allWorkEnumerated = false; + this.entityCounters = new ConcurrentHashMap<String, AtomicInteger>(); + this.synchronizerName = "Historical Entity Summarizer"; + this.enabledStatFlags = EnumSet.of(StatFlag.AAI_REST_STATS, StatFlag.ES_REST_STATS); + this.syncInProgress = false; + this.contextMap = MDC.getCopyOfContextMap(); + this.syncDurationInMs = -1; + this.searchableEntityLookup = searchableEntityLookup; + } + + /** + * Collect all the work. + * + * @return the operation state + */ + private OperationState collectAllTheWork() { + + Map<String, SearchableOxmEntityDescriptor> descriptorMap = + searchableEntityLookup.getSearchableEntityDescriptors(); + + if (descriptorMap.isEmpty()) { + LOG.error(AaiUiMsgs.OXM_FAILED_RETRIEVAL, "historical entities"); + + return OperationState.ERROR; + } + + Collection<String> entityTypes = descriptorMap.keySet(); + + AtomicInteger asyncWoH = new AtomicInteger(0); + + asyncWoH.set(entityTypes.size()); + + try { + for (String entityType : entityTypes) { + + supplyAsync(new Supplier<Void>() { + + @Override + public Void get() { + MDC.setContextMap(contextMap); + try { + OperationResult typeLinksResult = + aaiAdapter.getSelfLinksByEntityType(entityType); + updateActiveInventoryCounters(HttpMethod.GET, entityType, typeLinksResult); + processEntityTypeSelfLinks(entityType, typeLinksResult); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, exc.getMessage()); + + } + + return null; + } + + }, aaiExecutor).whenComplete((result, error) -> { + + asyncWoH.decrementAndGet(); + + if (error != null) { + LOG.error(AaiUiMsgs.HISTORICAL_COLLECT_ERROR, error.getMessage()); + } + + }); + + } + + + while (asyncWoH.get() > 0) { + + if (LOG.isDebugEnabled()) { + LOG.debug(AaiUiMsgs.DEBUG_GENERIC, indexName + " summarizer waiting for all the links to be processed."); + } + + Thread.sleep(250); + } + + esWorkOnHand.set(entityCounters.size()); + + // start doing the real work + allWorkEnumerated = true; + + insertEntityTypeCounters(); + + if (LOG.isDebugEnabled()) { + + StringBuilder sb = new StringBuilder(128); + + sb.append("\n\nHistorical Entity Counters:"); + + for (Entry<String, AtomicInteger> entry : entityCounters.entrySet()) { + sb.append("\n").append(entry.getKey()).append(" = ").append(entry.getValue().get()); + } + + LOG.debug(AaiUiMsgs.DEBUG_GENERIC, sb.toString()); + + } + + } catch (Exception exc) { + LOG.error(AaiUiMsgs.HISTORICAL_COLLECT_ERROR, exc.getMessage()); + + + esWorkOnHand.set(0); + allWorkEnumerated = true; + + return OperationState.ERROR; + } + + return OperationState.OK; + + } + + /* (non-Javadoc) + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync() + */ + @Override + public OperationState doSync() { + this.syncDurationInMs = -1; + String txnID = NodeUtils.getRandomTxnId(); + MdcContext.initialize(txnID, "HistoricalEntitySynchronizer", "", "Sync", ""); + + if (syncInProgress) { + LOG.info(AaiUiMsgs.HISTORICAL_SYNC_PENDING); + return OperationState.PENDING; + } + + clearCache(); + + syncInProgress = true; + this.syncStartedTimeStampInMs = System.currentTimeMillis(); + allWorkEnumerated = false; + + return collectAllTheWork(); + } + + /** + * Process entity type self links. + * + * @param entityType the entity type + * @param operationResult the operation result + */ + private void processEntityTypeSelfLinks(String entityType, OperationResult operationResult) { + + JsonNode rootNode = null; + + final String jsonResult = operationResult.getResult(); + + if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) { + + try { + rootNode = mapper.readTree(jsonResult); + } catch (IOException exc) { + LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, exc.getMessage()); + return; + } + + JsonNode resultData = rootNode.get("result-data"); + ArrayNode resultDataArrayNode = null; + + if (resultData != null && resultData.isArray()) { + resultDataArrayNode = (ArrayNode) resultData; + entityCounters.put(entityType, new AtomicInteger(resultDataArrayNode.size())); + } + } + + } + + /** + * Insert entity type counters. + */ + private void insertEntityTypeCounters() { + + if (esWorkOnHand.get() <= 0) { + return; + } + + SimpleDateFormat dateFormat = new SimpleDateFormat(INSERTION_DATE_TIME_FORMAT); + Timestamp timestamp = new Timestamp(System.currentTimeMillis()); + String currentFormattedTimeStamp = dateFormat.format(timestamp); + + Set<Entry<String, AtomicInteger>> entityCounterEntries = entityCounters.entrySet(); + + for (Entry<String, AtomicInteger> entityCounterEntry : entityCounterEntries) { + + supplyAsync(new Supplier<Void>() { + + @Override + public Void get() { + MDC.setContextMap(contextMap); + String jsonString = Json.createObjectBuilder().add( + "count", entityCounterEntry.getValue().get()) + .add("entityType", entityCounterEntry.getKey()) + .add("timestamp", currentFormattedTimeStamp).build().toString(); + + String link = null; + try { + link = elasticSearchAdapter.buildElasticSearchPostUrl(indexName); + OperationResult or = elasticSearchAdapter.doPost(link, jsonString, MediaType.APPLICATION_JSON_TYPE); + updateElasticSearchCounters(HttpMethod.POST, entityCounterEntry.getKey(), or); + } catch (Exception exc) { + LOG.error(AaiUiMsgs.ES_STORE_FAILURE, exc.getMessage() ); + } + + return null; + } + + }, esExecutor).whenComplete((result, error) -> { + + esWorkOnHand.decrementAndGet(); + + }); + + } + + while (esWorkOnHand.get() > 0) { + + try { + Thread.sleep(500); + } catch (InterruptedException exc) { + LOG.error(AaiUiMsgs.INTERRUPTED, "historical Entities", exc.getMessage()); + } + } + + } + + @Override + public SynchronizerState getState() { + + if (!isSyncDone()) { + return SynchronizerState.PERFORMING_SYNCHRONIZATION; + } + + return SynchronizerState.IDLE; + + } + + /* (non-Javadoc) + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean) + */ + @Override + public String getStatReport(boolean showFinalReport) { + syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs; + return this.getStatReport(syncDurationInMs, showFinalReport); + } + + /* (non-Javadoc) + * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown() + */ + @Override + public void shutdown() { + this.shutdownExecutors(); + } + + @Override + protected boolean isSyncDone() { + + int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get(); + + if (LOG.isDebugEnabled()) { + LOG.debug(AaiUiMsgs.DEBUG_GENERIC,indexName + ", isSyncDone(), totalWorkOnHand = " + totalWorkOnHand + + " all work enumerated = " + allWorkEnumerated); + } + + if (totalWorkOnHand > 0 || !allWorkEnumerated) { + return false; + } + + this.syncInProgress = false; + + return true; + } + + /* (non-Javadoc) + * @see org.openecomp.sparky.synchronizer.AbstractEntitySynchronizer#clearCache() + */ + @Override + public void clearCache() { + + if (syncInProgress) { + LOG.debug(AaiUiMsgs.DEBUG_GENERIC, "Historical Entity Summarizer in progress, request to clear cache ignored"); + return; + } + + super.clearCache(); + this.resetCounters(); + if (entityCounters != null) { + entityCounters.clear(); + } + + allWorkEnumerated = false; + + } + +} diff --git a/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySyncController.java b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySyncController.java new file mode 100644 index 0000000..eb42489 --- /dev/null +++ b/sparkybe-onap-service/src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySyncController.java @@ -0,0 +1,94 @@ +/** + * ============LICENSE_START=================================================== + * SPARKY (AAI UI service) + * ============================================================================ + * Copyright © 2017 AT&T Intellectual Property. + * Copyright © 2017 Amdocs + * All rights reserved. + * ============================================================================ + * 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 and OpenECOMP are trademarks + * and service marks of AT&T Intellectual Property. + */ +package org.onap.aai.sparky.aggregation.sync; + +import org.onap.aai.sparky.config.oxm.SearchableEntityLookup; +import org.onap.aai.sparky.dal.ActiveInventoryAdapter; +import org.onap.aai.sparky.dal.ElasticSearchAdapter; +import org.onap.aai.sparky.sync.ElasticSearchSchemaFactory; +import org.onap.aai.sparky.sync.IndexIntegrityValidator; +import org.onap.aai.sparky.sync.SyncControllerImpl; +import org.onap.aai.sparky.sync.SyncControllerRegistrar; +import org.onap.aai.sparky.sync.SyncControllerRegistry; +import org.onap.aai.sparky.sync.config.ElasticSearchEndpointConfig; +import org.onap.aai.sparky.sync.config.ElasticSearchSchemaConfig; +import org.onap.aai.sparky.sync.config.NetworkStatisticsConfig; +import org.onap.aai.sparky.sync.config.SyncControllerConfig; + +public class HistoricalEntitySyncController extends SyncControllerImpl + implements SyncControllerRegistrar { + + private SyncControllerRegistry syncControllerRegistry; + + public HistoricalEntitySyncController(SyncControllerConfig syncControllerConfig, + ActiveInventoryAdapter aaiAdapter, ElasticSearchAdapter esAdapter, + ElasticSearchSchemaConfig schemaConfig, ElasticSearchEndpointConfig endpointConfig, + int syncFrequencyInMinutes, NetworkStatisticsConfig aaiStatConfig, + NetworkStatisticsConfig esStatConfig, SearchableEntityLookup searchableEntityLookup, + ElasticSearchSchemaFactory elasticSearchSchemaFactory) throws Exception { + super(syncControllerConfig); + + // final String controllerName = "Historical Entity Count Synchronizer"; + + long taskFrequencyInMs = syncFrequencyInMinutes * 60 * 1000; + + setDelayInMs(taskFrequencyInMs); + setSyncFrequencyInMs(taskFrequencyInMs); + + IndexIntegrityValidator entityCounterHistoryValidator = new IndexIntegrityValidator(esAdapter, + schemaConfig, endpointConfig, elasticSearchSchemaFactory.getIndexSchema(schemaConfig)); + + registerIndexValidator(entityCounterHistoryValidator); + + HistoricalEntitySummarizer historicalSummarizer = new HistoricalEntitySummarizer(schemaConfig, + syncControllerConfig.getNumInternalSyncWorkers(), + syncControllerConfig.getNumSyncActiveInventoryWorkers(), + syncControllerConfig.getNumSyncElasticWorkers(),aaiStatConfig, esStatConfig,searchableEntityLookup); + + historicalSummarizer.setAaiAdapter(aaiAdapter); + historicalSummarizer.setElasticSearchAdapter(esAdapter); + + registerEntitySynchronizer(historicalSummarizer); + + } + + public SyncControllerRegistry getSyncControllerRegistry() { + return syncControllerRegistry; + } + + public void setSyncControllerRegistry(SyncControllerRegistry syncControllerRegistry) { + this.syncControllerRegistry = syncControllerRegistry; + } + + @Override + public void registerController() { + if ( syncControllerRegistry != null ) { + if ( syncControllerConfig.isEnabled()) { + syncControllerRegistry.registerSyncController(this); + } + } + + } +} |