src/main/java/org/onap/aai/sparky/synchronizer/AggregationSuggestionSynchronizer.java

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * org.onap.aai
   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
  11  *
  12  *       http://www.apache.org/licenses/LICENSE-2.0
  13  *
  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=========================================================
  20  *
  21  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  */
  23 package org.onap.aai.sparky.synchronizer;
  24
  25 import static java.util.concurrent.CompletableFuture.supplyAsync;
  26
  27 import java.util.Map;
  28 import java.util.concurrent.ExecutorService;
  29
  30 import org.onap.aai.sparky.dal.NetworkTransaction;
  31 import org.onap.aai.sparky.dal.rest.HttpMethod;
  32 import org.onap.aai.sparky.dal.rest.OperationResult;
  33 import org.onap.aai.sparky.logging.AaiUiMsgs;
  34 import org.onap.aai.sparky.synchronizer.entity.AggregationSuggestionEntity;
  35 import org.onap.aai.sparky.synchronizer.enumeration.OperationState;
  36 import org.onap.aai.sparky.synchronizer.enumeration.SynchronizerState;
  37 import org.onap.aai.sparky.synchronizer.task.PerformElasticSearchPut;
  38 import org.onap.aai.sparky.util.NodeUtils;
  39 import org.onap.aai.cl.api.Logger;
  40 import org.onap.aai.cl.eelf.LoggerFactory;
  41 import org.onap.aai.cl.mdc.MdcContext;
  42 import org.slf4j.MDC;
  43
  44 public class AggregationSuggestionSynchronizer extends AbstractEntitySynchronizer
  45     implements IndexSynchronizer {
  46
  47   private static final Logger LOG =
  48       LoggerFactory.getInstance().getLogger(AggregationSuggestionSynchronizer.class);
  49
  50   private boolean isSyncInProgress;
  51   private boolean shouldPerformRetry;
  52   private Map<String, String> contextMap;
  53   protected ExecutorService esPutExecutor;
  54
  55   public AggregationSuggestionSynchronizer(String indexName) throws Exception {
  56     super(LOG, "ASS-" + indexName.toUpperCase(), 2, 5, 5, indexName);
  57
  58     this.isSyncInProgress = false;
  59     this.shouldPerformRetry = false;
  60     this.synchronizerName = "Aggregation Suggestion Synchronizer";
  61     this.contextMap = MDC.getCopyOfContextMap();
  62     this.esPutExecutor = NodeUtils.createNamedExecutor("ASS-ES-PUT", 2, LOG);
  63   }
  64
  65   @Override
  66   protected boolean isSyncDone() {
  67     int totalWorkOnHand = esWorkOnHand.get();
  68
  69     if (LOG.isDebugEnabled()) {
  70       LOG.debug(AaiUiMsgs.DEBUG_GENERIC,
  71           indexName + ", isSyncDone(), totalWorkOnHand = " + totalWorkOnHand);
  72     }
  73
  74     if (totalWorkOnHand > 0 || !isSyncInProgress) {
  75       return false;
  76     }
  77
  78     return true;
  79   }
  80
  81   @Override
  82   public OperationState doSync() {
  83     isSyncInProgress = true;
  84     this.syncDurationInMs = -1;
  85     syncStartedTimeStampInMs = System.currentTimeMillis();
  86     syncEntity();
  87
  88     while (!isSyncDone()) {
  89       try {
  90         if (shouldPerformRetry) {
  91           syncEntity();
  92         }
  93         Thread.sleep(1000);
  94       } catch (Exception exc) {
  95         // We don't care about this exception
  96       }
  97     }
  98
  99     return OperationState.OK;
 100   }
 101
 102   private void syncEntity() {
 103     String txnId = NodeUtils.getRandomTxnId();
 104     MdcContext.initialize(txnId, "AggregationSuggestionSynchronizer", "", "Sync", "");
 105
 106     AggregationSuggestionEntity syncEntity = new AggregationSuggestionEntity();
 107     syncEntity.deriveFields();
 108
 109     String link = null;
 110     try {
 111       link = getElasticFullUrl("/" + syncEntity.getId(), getIndexName());
 112     } catch (Exception exc) {
 113       LOG.error(AaiUiMsgs.ES_LINK_UPSERT, exc.getLocalizedMessage());
 114     }
 115
 116     try {
 117       String jsonPayload = null;
 118       jsonPayload = syncEntity.getIndexDocumentJson();
 119       if (link != null && jsonPayload != null) {
 120
 121         NetworkTransaction elasticPutTxn = new NetworkTransaction();
 122         elasticPutTxn.setLink(link);
 123         elasticPutTxn.setOperationType(HttpMethod.PUT);
 124
 125         esWorkOnHand.incrementAndGet();
 126         final Map<String, String> contextMap = MDC.getCopyOfContextMap();
 127         supplyAsync(new PerformElasticSearchPut(jsonPayload, elasticPutTxn,
 128             esDataProvider, contextMap), esPutExecutor).whenComplete((result, error) -> {
 129
 130               esWorkOnHand.decrementAndGet();
 131
 132               if (error != null) {
 133                 String message = "Aggregation suggestion entity sync UPDATE PUT error - "
 134                     + error.getLocalizedMessage();
 135                 LOG.error(AaiUiMsgs.ES_AGGREGATION_SUGGESTION_ENTITY_SYNC_ERROR, message);
 136               } else {
 137                 updateElasticSearchCounters(result);
 138                 wasEsOperationSuccessful(result);
 139               }
 140             });
 141       }
 142     } catch (Exception exc) {
 143       String message =
 144           "Exception caught during aggregation suggestion entity sync PUT operation. Message - "
 145               + exc.getLocalizedMessage();
 146       LOG.error(AaiUiMsgs.ES_AGGREGATION_SUGGESTION_ENTITY_SYNC_ERROR, message);
 147     }
 148   }
 149
 150   private void wasEsOperationSuccessful(NetworkTransaction result) {
 151     if (result != null) {
 152       OperationResult opResult = result.getOperationResult();
 153
 154       if (!opResult.wasSuccessful()) {
 155         shouldPerformRetry = true;
 156       } else {
 157         isSyncInProgress = false;
 158         shouldPerformRetry = false;
 159       }
 160     }
 161   }
 162
 163   @Override
 164   public SynchronizerState getState() {
 165     if (!isSyncDone()) {
 166       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 167     }
 168
 169     return SynchronizerState.IDLE;
 170   }
 171
 172   @Override
 173   public String getStatReport(boolean shouldDisplayFinalReport) {
 174         syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 175         return getStatReport(syncDurationInMs, shouldDisplayFinalReport);
 176   }
 177
 178   @Override
 179   public void shutdown() {
 180     this.shutdownExecutors();
 181   }
 182 }