+++ /dev/null
-/**
- * ============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.openecomp.sparky.synchronizer;
-
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.EnumSet;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.openecomp.cl.api.Logger;
-import org.openecomp.sparky.config.oxm.OxmModelLoader;
-import org.openecomp.sparky.dal.NetworkTransaction;
-import org.openecomp.sparky.dal.aai.ActiveInventoryDataProvider;
-import org.openecomp.sparky.dal.aai.ActiveInventoryEntityStatistics;
-import org.openecomp.sparky.dal.aai.ActiveInventoryProcessingExceptionStatistics;
-import org.openecomp.sparky.dal.aai.config.ActiveInventoryConfig;
-import org.openecomp.sparky.dal.elasticsearch.ElasticSearchDataProvider;
-import org.openecomp.sparky.dal.elasticsearch.ElasticSearchEntityStatistics;
-import org.openecomp.sparky.dal.elasticsearch.config.ElasticSearchConfig;
-import org.openecomp.sparky.dal.rest.HttpMethod;
-import org.openecomp.sparky.dal.rest.OperationResult;
-import org.openecomp.sparky.dal.rest.RestOperationalStatistics;
-import org.openecomp.sparky.logging.AaiUiMsgs;
-import org.openecomp.sparky.util.NodeUtils;
-
-import org.openecomp.cl.mdc.MdcContext;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-/**
- * The Class AbstractEntitySynchronizer.
- *
- * @author davea.
- */
-public abstract class AbstractEntitySynchronizer {
-
- protected static final int VERSION_CONFLICT_EXCEPTION_CODE = 409;
- protected static final Integer RETRY_COUNT_PER_ENTITY_LIMIT = new Integer(3);
-
- protected final Logger logger;
- protected ObjectMapper mapper;
- protected OxmModelLoader oxmModelLoader;
- protected long syncDurationInMs;
- /**
- * The Enum StatFlag.
- */
- protected enum StatFlag {
- AAI_REST_STATS, AAI_ENTITY_STATS, AAI_PROCESSING_EXCEPTION_STATS,
- AAI_TASK_PROCESSING_STATS, ES_REST_STATS, ES_ENTITY_STATS, ES_TASK_PROCESSING_STATS
- }
-
- protected EnumSet<StatFlag> enabledStatFlags;
-
- protected ActiveInventoryDataProvider aaiDataProvider;
- protected ElasticSearchDataProvider esDataProvider;
-
- protected ExecutorService synchronizerExecutor;
- protected ExecutorService aaiExecutor;
- protected ExecutorService esExecutor;
-
- private RestOperationalStatistics esRestStats;
- protected ElasticSearchEntityStatistics esEntityStats;
-
- private RestOperationalStatistics aaiRestStats;
- protected ActiveInventoryEntityStatistics aaiEntityStats;
- private ActiveInventoryProcessingExceptionStatistics aaiProcessingExceptionStats;
-
- private TaskProcessingStats aaiTaskProcessingStats;
- private TaskProcessingStats esTaskProcessingStats;
-
- private TransactionRateController aaiTransactionRateController;
- private TransactionRateController esTransactionRateController;
-
- protected AtomicInteger aaiWorkOnHand;
- protected AtomicInteger esWorkOnHand;
- protected String synchronizerName;
-
- protected abstract boolean isSyncDone();
- protected boolean shouldSkipSync;
-
- public String getActiveInventoryStatisticsReport() {
-
- StringBuilder sb = new StringBuilder(128);
-
- if (enabledStatFlags.contains(StatFlag.AAI_REST_STATS)) {
- sb.append("\n\n ").append("REST Operational Stats:");
- sb.append(aaiRestStats.getStatisticsReport());
- }
-
- if (enabledStatFlags.contains(StatFlag.AAI_ENTITY_STATS)) {
- sb.append("\n\n ").append("Entity Stats:");
- sb.append(aaiEntityStats.getStatisticsReport());
- }
-
- if (enabledStatFlags.contains(StatFlag.AAI_PROCESSING_EXCEPTION_STATS)) {
- sb.append("\n\n ").append("Processing Exception Stats:");
- sb.append(aaiProcessingExceptionStats.getStatisticsReport());
- }
-
- return sb.toString();
-
- }
-
- public String getElasticSearchStatisticsReport() {
-
- StringBuilder sb = new StringBuilder(128);
-
- if (enabledStatFlags.contains(StatFlag.ES_REST_STATS)) {
- sb.append("\n\n ").append("REST Operational Stats:");
- sb.append(esRestStats.getStatisticsReport());
- }
-
- if (enabledStatFlags.contains(StatFlag.ES_ENTITY_STATS)) {
- sb.append("\n\n ").append("Entity Stats:");
- sb.append(esEntityStats.getStatisticsReport());
- }
-
- return sb.toString();
-
- }
-
- /**
- * Adds the active inventory stat report.
- *
- * @param sb the sb
- */
- private void addActiveInventoryStatReport(StringBuilder sb) {
-
- if (sb == null) {
- return;
- }
-
- sb.append("\n\n AAI");
- sb.append(getActiveInventoryStatisticsReport());
-
- double currentTps = 0;
- if (enabledStatFlags.contains(StatFlag.AAI_TASK_PROCESSING_STATS)) {
- sb.append("\n\n ").append("Task Processor Stats:");
- sb.append(aaiTaskProcessingStats.getStatisticsReport(false, " "));
-
- currentTps = aaiTransactionRateController.getCurrentTps();
-
- sb.append("\n ").append("Current TPS: ").append(currentTps);
- }
-
- sb.append("\n ").append("Current WOH: ").append(aaiWorkOnHand.get());
-
- if (enabledStatFlags.contains(StatFlag.AAI_TASK_PROCESSING_STATS)) {
- if (currentTps > 0) {
- double numMillisecondsToCompletion = (aaiWorkOnHand.get() / currentTps) * 1000;
- sb.append("\n ").append("SyncDurationRemaining=")
- .append(NodeUtils.getDurationBreakdown((long) numMillisecondsToCompletion));
- }
- }
-
- }
-
- /**
- * Adds the elastic stat report.
- *
- * @param sb the sb
- */
- private void addElasticStatReport(StringBuilder sb) {
-
- if (sb == null) {
- return;
- }
-
- sb.append("\n\n ELASTIC");
- sb.append(getElasticSearchStatisticsReport());
-
- double currentTps = 0;
-
- if (enabledStatFlags.contains(StatFlag.ES_TASK_PROCESSING_STATS)) {
- sb.append("\n\n ").append("Task Processor Stats:");
- sb.append(esTaskProcessingStats.getStatisticsReport(false, " "));
-
- currentTps = esTransactionRateController.getCurrentTps();
-
- sb.append("\n ").append("Current TPS: ").append(currentTps);
- }
-
- sb.append("\n ").append("Current WOH: ").append(esWorkOnHand.get());
-
- if (enabledStatFlags.contains(StatFlag.ES_TASK_PROCESSING_STATS)) {
- if (currentTps > 0) {
- double numMillisecondsToCompletion = (esWorkOnHand.get() / currentTps) * 1000;
- sb.append("\n ").append("SyncDurationRemaining=")
- .append(NodeUtils.getDurationBreakdown((long) numMillisecondsToCompletion));
- }
- }
-
-
- }
-
- /**
- * Gets the stat report.
- *
- * @param syncOpTimeInMs the sync op time in ms
- * @param showFinalReport the show final report
- * @return the stat report
- */
- protected String getStatReport(long syncOpTimeInMs, boolean showFinalReport) {
-
- StringBuilder sb = new StringBuilder(128);
-
- sb.append("\n").append(synchronizerName + " Statistics: ( Sync Operation Duration = "
- + NodeUtils.getDurationBreakdown(syncOpTimeInMs) + " )");
-
- addActiveInventoryStatReport(sb);
- addElasticStatReport(sb);
-
- if (showFinalReport) {
- sb.append("\n\n ").append("Sync Completed!\n");
- } else {
- sb.append("\n\n ").append("Sync in Progress...\n");
- }
-
- return sb.toString();
-
- }
-
- protected String indexName;
- protected long syncStartedTimeStampInMs;
-
- /**
- * Instantiates a new abstract entity synchronizer.
- *
- * @param logger the logger
- * @param syncName the sync name
- * @param numSyncWorkers the num sync workers
- * @param numActiveInventoryWorkers the num active inventory workers
- * @param numElasticsearchWorkers the num elasticsearch workers
- * @param indexName the index name
- * @throws Exception the exception
- */
- protected AbstractEntitySynchronizer(Logger logger, String syncName, int numSyncWorkers,
- int numActiveInventoryWorkers, int numElasticsearchWorkers, String indexName)
- throws Exception {
- this.logger = logger;
- this.synchronizerExecutor =
- NodeUtils.createNamedExecutor(syncName + "-INTERNAL", numSyncWorkers, logger);
- this.aaiExecutor =
- NodeUtils.createNamedExecutor(syncName + "-AAI", numActiveInventoryWorkers, logger);
- this.esExecutor =
- NodeUtils.createNamedExecutor(syncName + "-ES", numElasticsearchWorkers, logger);
- this.mapper = new ObjectMapper();
- this.oxmModelLoader = OxmModelLoader.getInstance();
- this.indexName = indexName;
- this.esRestStats = new RestOperationalStatistics();
- this.esEntityStats = new ElasticSearchEntityStatistics(oxmModelLoader);
- this.aaiRestStats = new RestOperationalStatistics();
- this.aaiEntityStats = new ActiveInventoryEntityStatistics(oxmModelLoader);
- this.aaiProcessingExceptionStats = new ActiveInventoryProcessingExceptionStatistics();
- this.aaiTaskProcessingStats =
- new TaskProcessingStats(ActiveInventoryConfig.getConfig().getTaskProcessorConfig());
- this.esTaskProcessingStats =
- new TaskProcessingStats(ElasticSearchConfig.getConfig().getProcessorConfig());
-
- this.aaiTransactionRateController =
- new TransactionRateController(ActiveInventoryConfig.getConfig().getTaskProcessorConfig());
- this.esTransactionRateController =
- new TransactionRateController(ElasticSearchConfig.getConfig().getProcessorConfig());
-
- this.aaiWorkOnHand = new AtomicInteger(0);
- this.esWorkOnHand = new AtomicInteger(0);
-
- enabledStatFlags = EnumSet.allOf(StatFlag.class);
-
- this.synchronizerName = "Abstact Entity Synchronizer";
-
- String txnID = NodeUtils.getRandomTxnId();
- MdcContext.initialize(txnID, "AbstractEntitySynchronizer", "", "Sync", "");
-
- this.shouldSkipSync = false;
- this.syncStartedTimeStampInMs = System.currentTimeMillis();
- this.syncDurationInMs = -1;
- }
-
- public boolean shouldSkipSync() {
- return shouldSkipSync;
- }
-
- public void setShouldSkipSync(boolean shouldSkipSync) {
- this.shouldSkipSync = shouldSkipSync;
- }
-
- /**
- * Inc active inventory work on hand counter.
- */
- protected void incActiveInventoryWorkOnHandCounter() {
- aaiWorkOnHand.incrementAndGet();
- }
-
- /**
- * Dec active inventory work on hand counter.
- */
- protected void decActiveInventoryWorkOnHandCounter() {
- aaiWorkOnHand.decrementAndGet();
- }
-
- /**
- * Inc elastic search work on hand counter.
- */
- protected void incElasticSearchWorkOnHandCounter() {
- esWorkOnHand.incrementAndGet();
- }
-
- /**
- * Dec elastic search work on hand counter.
- */
- protected void decElasticSearchWorkOnHandCounter() {
- esWorkOnHand.decrementAndGet();
- }
-
- /**
- * Shutdown executors.
- */
- protected void shutdownExecutors() {
- try {
- synchronizerExecutor.shutdown();
- aaiExecutor.shutdown();
- esExecutor.shutdown();
- aaiDataProvider.shutdown();
- esDataProvider.shutdown();
- } catch (Exception exc) {
- logger.error(AaiUiMsgs.ERROR_SHUTDOWN_EXECUTORS, exc );
- }
- }
-
- /**
- * Clear cache.
- */
- public void clearCache() {
- if (aaiDataProvider != null) {
- aaiDataProvider.clearCache();
- }
- }
-
- protected ActiveInventoryDataProvider getAaiDataProvider() {
- return aaiDataProvider;
- }
-
- public void setAaiDataProvider(ActiveInventoryDataProvider aaiDataProvider) {
- this.aaiDataProvider = aaiDataProvider;
- }
-
- protected ElasticSearchDataProvider getEsDataProvider() {
- return esDataProvider;
- }
-
- public void setEsDataProvider(ElasticSearchDataProvider provider) {
- this.esDataProvider = provider;
- }
-
- /**
- * Gets the elastic full url.
- *
- * @param resourceUrl the resource url
- * @param indexName the index name
- * @param indexType the index type
- * @return the elastic full url
- * @throws Exception the exception
- */
- protected String getElasticFullUrl(String resourceUrl, String indexName, String indexType)
- throws Exception {
- return ElasticSearchConfig.getConfig().getElasticFullUrl(resourceUrl, indexName, indexType);
- }
-
- /**
- * Gets the elastic full url.
- *
- * @param resourceUrl the resource url
- * @param indexName the index name
- * @return the elastic full url
- * @throws Exception the exception
- */
- protected String getElasticFullUrl(String resourceUrl, String indexName) throws Exception {
- return ElasticSearchConfig.getConfig().getElasticFullUrl(resourceUrl, indexName);
- }
-
- public String getIndexName() {
- return indexName;
- }
-
- public void setIndexName(String indexName) {
- this.indexName = indexName;
- }
-
-
- /**
- * Gets the response length.
- *
- * @param txn the txn
- * @return the response length
- */
- private long getResponseLength(NetworkTransaction txn) {
-
- if (txn == null) {
- return -1;
- }
-
- OperationResult result = txn.getOperationResult();
-
- if (result == null) {
- return -1;
- }
-
- if (result.getResult() != null) {
- return result.getResult().length();
- }
-
- return -1;
- }
-
- /**
- * Update elastic search counters.
- *
- * @param method the method
- * @param or the or
- */
- protected void updateElasticSearchCounters(HttpMethod method, OperationResult or) {
- updateElasticSearchCounters(new NetworkTransaction(method, null, or));
- }
-
- /**
- * Update elastic search counters.
- *
- * @param method the method
- * @param entityType the entity type
- * @param or the or
- */
- protected void updateElasticSearchCounters(HttpMethod method, String entityType,
- OperationResult or) {
- updateElasticSearchCounters(new NetworkTransaction(method, entityType, or));
- }
-
- /**
- * Update elastic search counters.
- *
- * @param txn the txn
- */
- protected void updateElasticSearchCounters(NetworkTransaction txn) {
-
- if (enabledStatFlags.contains(StatFlag.ES_REST_STATS)) {
- esRestStats.updateCounters(txn);
- }
-
- if (enabledStatFlags.contains(StatFlag.ES_ENTITY_STATS)) {
- esEntityStats.updateCounters(txn);
- }
-
- if (enabledStatFlags.contains(StatFlag.ES_TASK_PROCESSING_STATS)) {
-
- esTransactionRateController.trackResponseTime(txn.getOperationResult().getResponseTimeInMs());
-
- esTaskProcessingStats
- .updateTaskResponseStatsHistogram(txn.getOperationResult().getResponseTimeInMs());
- esTaskProcessingStats.updateTaskAgeStatsHistogram(txn.getTaskAgeInMs());
-
- // don't know the cost of the lengh calc, we'll see if it causes a
- // problem
-
- long responsePayloadSizeInBytes = getResponseLength(txn);
- if (responsePayloadSizeInBytes >= 0) {
- esTaskProcessingStats.updateResponseSizeInBytesHistogram(responsePayloadSizeInBytes);
- }
-
- esTaskProcessingStats
- .updateTransactionsPerSecondHistogram((long) esTransactionRateController.getCurrentTps());
- }
- }
-
- /**
- * Update active inventory counters.
- *
- * @param method the method
- * @param or the or
- */
- protected void updateActiveInventoryCounters(HttpMethod method, OperationResult or) {
- updateActiveInventoryCounters(new NetworkTransaction(method, null, or));
- }
-
- /**
- * Update active inventory counters.
- *
- * @param method the method
- * @param entityType the entity type
- * @param or the or
- */
- protected void updateActiveInventoryCounters(HttpMethod method, String entityType,
- OperationResult or) {
- updateActiveInventoryCounters(new NetworkTransaction(method, entityType, or));
- }
-
- /**
- * Update active inventory counters.
- *
- * @param txn the txn
- */
- protected void updateActiveInventoryCounters(NetworkTransaction txn) {
-
- if (enabledStatFlags.contains(StatFlag.AAI_REST_STATS)) {
- aaiRestStats.updateCounters(txn);
- }
-
- if (enabledStatFlags.contains(StatFlag.AAI_ENTITY_STATS)) {
- aaiEntityStats.updateCounters(txn);
- }
-
- if (enabledStatFlags.contains(StatFlag.AAI_PROCESSING_EXCEPTION_STATS)) {
- aaiProcessingExceptionStats.updateCounters(txn);
- }
-
- if (enabledStatFlags.contains(StatFlag.AAI_TASK_PROCESSING_STATS)) {
- aaiTransactionRateController
- .trackResponseTime(txn.getOperationResult().getResponseTimeInMs());
-
- aaiTaskProcessingStats
- .updateTaskResponseStatsHistogram(txn.getOperationResult().getResponseTimeInMs());
- aaiTaskProcessingStats.updateTaskAgeStatsHistogram(txn.getTaskAgeInMs());
-
- // don't know the cost of the lengh calc, we'll see if it causes a
- // problem
-
- long responsePayloadSizeInBytes = getResponseLength(txn);
- if (responsePayloadSizeInBytes >= 0) {
- aaiTaskProcessingStats.updateResponseSizeInBytesHistogram(responsePayloadSizeInBytes);
- }
-
- aaiTaskProcessingStats.updateTransactionsPerSecondHistogram(
- (long) aaiTransactionRateController.getCurrentTps());
- }
- }
-
- /**
- * Reset counters.
- */
- protected void resetCounters() {
- aaiRestStats.reset();
- aaiEntityStats.reset();
- aaiProcessingExceptionStats.reset();
-
- esRestStats.reset();
- esEntityStats.reset();
- }
-
-}
Code Review / aai / sparky-be.git / blobdiff