src/main/java/org/openecomp/sparky/synchronizer/SearchableEntitySynchronizer.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.openecomp.sparky.synchronizer;
  24
  25 import static java.util.concurrent.CompletableFuture.supplyAsync;
  26
  27 import org.openecomp.cl.mdc.MdcContext;
  28
  29 import org.openecomp.cl.mdc.MdcContext;
  30
  31 import com.fasterxml.jackson.core.JsonProcessingException;
  32 import com.fasterxml.jackson.databind.JsonNode;
  33 import com.fasterxml.jackson.databind.ObjectReader;
  34 import com.fasterxml.jackson.databind.node.ArrayNode;
  35
  36 import java.io.IOException;
  37 import java.net.InetAddress;
  38 import java.net.UnknownHostException;
  39 import java.util.ArrayList;
  40 import java.util.Collection;
  41 import java.util.Deque;
  42 import java.util.Iterator;
  43 import java.util.List;
  44 import java.util.Map;
  45 import java.util.concurrent.ConcurrentHashMap;
  46 import java.util.concurrent.ConcurrentLinkedDeque;
  47 import java.util.concurrent.ExecutorService;
  48 import java.util.function.Supplier;
  49
  50 import org.openecomp.cl.api.Logger;
  51 import org.openecomp.cl.eelf.LoggerFactory;
  52 import org.openecomp.sparky.config.oxm.OxmEntityDescriptor;
  53 import org.openecomp.sparky.dal.NetworkTransaction;
  54 import org.openecomp.sparky.dal.aai.config.ActiveInventoryConfig;
  55 import org.openecomp.sparky.dal.elasticsearch.config.ElasticSearchConfig;
  56 import org.openecomp.sparky.dal.rest.HttpMethod;
  57 import org.openecomp.sparky.dal.rest.OperationResult;
  58 import org.openecomp.sparky.logging.AaiUiMsgs;
  59 import org.openecomp.sparky.synchronizer.config.SynchronizerConfiguration;
  60 import org.openecomp.sparky.synchronizer.entity.MergableEntity;
  61 import org.openecomp.sparky.synchronizer.entity.SearchableEntity;
  62 import org.openecomp.sparky.synchronizer.entity.SelfLinkDescriptor;
  63 import org.openecomp.sparky.synchronizer.enumeration.OperationState;
  64 import org.openecomp.sparky.synchronizer.enumeration.SynchronizerState;
  65 import org.openecomp.sparky.synchronizer.task.PerformActiveInventoryRetrieval;
  66 import org.openecomp.sparky.synchronizer.task.PerformElasticSearchPut;
  67 import org.openecomp.sparky.synchronizer.task.PerformElasticSearchRetrieval;
  68 import org.openecomp.sparky.synchronizer.task.PerformElasticSearchUpdate;
  69 import org.openecomp.sparky.util.NodeUtils;
  70 import org.slf4j.MDC;
  71
  72 /**
  73  * The Class SearchableEntitySynchronizer.
  74  */
  75 public class SearchableEntitySynchronizer extends AbstractEntitySynchronizer
  76     implements IndexSynchronizer {
  77
  78   /**
  79    * The Class RetrySearchableEntitySyncContainer.
  80    */
  81   private class RetrySearchableEntitySyncContainer {
  82     NetworkTransaction txn;
  83     SearchableEntity se;
  84
  85     /**
  86      * Instantiates a new retry searchable entity sync container.
  87      *
  88      * @param txn the txn
  89      * @param se the se
  90      */
  91     public RetrySearchableEntitySyncContainer(NetworkTransaction txn, SearchableEntity se) {
  92       this.txn = txn;
  93       this.se = se;
  94     }
  95
  96     public NetworkTransaction getNetworkTransaction() {
  97       return txn;
  98     }
  99
 100     public SearchableEntity getSearchableEntity() {
 101       return se;
 102     }
 103   }
 104
 105   private static final Logger LOG =
 106       LoggerFactory.getInstance().getLogger(SearchableEntitySynchronizer.class);
 107
 108   private boolean allWorkEnumerated;
 109   private Deque<SelfLinkDescriptor> selflinks;
 110   private Deque<RetrySearchableEntitySyncContainer> retryQueue;
 111   private Map<String, Integer> retryLimitTracker;
 112   protected ExecutorService esPutExecutor;
 113
 114   /**
 115    * Instantiates a new searchable entity synchronizer.
 116    *
 117    * @param indexName the index name
 118    * @throws Exception the exception
 119    */
 120   public SearchableEntitySynchronizer(String indexName) throws Exception {
 121     super(LOG, "SES", 2, 5, 5, indexName);
 122     this.allWorkEnumerated = false;
 123     this.selflinks = new ConcurrentLinkedDeque<SelfLinkDescriptor>();
 124     this.retryQueue = new ConcurrentLinkedDeque<RetrySearchableEntitySyncContainer>();
 125     this.retryLimitTracker = new ConcurrentHashMap<String, Integer>();
 126     this.synchronizerName = "Searchable Entity Synchronizer";
 127     this.esPutExecutor = NodeUtils.createNamedExecutor("SES-ES-PUT", 5, LOG);
 128     this.aaiEntityStats.initializeCountersFromOxmEntityDescriptors(
 129         oxmModelLoader.getSearchableEntityDescriptors());
 130     this.esEntityStats.initializeCountersFromOxmEntityDescriptors(
 131         oxmModelLoader.getSearchableEntityDescriptors());
 132     this.syncDurationInMs = -1;
 133   }
 134
 135   /**
 136    * Collect all the work.
 137    *
 138    * @return the operation state
 139    */
 140   private OperationState collectAllTheWork() {
 141     final Map<String, String> contextMap = MDC.getCopyOfContextMap();
 142     Map<String, OxmEntityDescriptor> descriptorMap =
 143         oxmModelLoader.getSearchableEntityDescriptors();
 144
 145     if (descriptorMap.isEmpty()) {
 146       LOG.error(AaiUiMsgs.ERROR_LOADING_OXM_SEARCHABLE_ENTITIES);
 147       LOG.info(AaiUiMsgs.ERROR_LOADING_OXM_SEARCHABLE_ENTITIES);
 148       return OperationState.ERROR;
 149     }
 150
 151     Collection<String> syncTypes = descriptorMap.keySet();
 152
 153     /*Collection<String> syncTypes = new ArrayList<String>();
 154     syncTypes.add("service-instance");*/
 155
 156     try {
 157
 158       /*
 159        * launch a parallel async thread to process the documents for each entity-type (to max the
 160        * of the configured executor anyway)
 161        */
 162
 163       aaiWorkOnHand.set(syncTypes.size());
 164
 165       for (String key : syncTypes) {
 166
 167         supplyAsync(new Supplier<Void>() {
 168
 169           @Override
 170           public Void get() {
 171             MDC.setContextMap(contextMap);
 172             OperationResult typeLinksResult = null;
 173             try {
 174               typeLinksResult = aaiDataProvider.getSelfLinksByEntityType(key);
 175               aaiWorkOnHand.decrementAndGet();
 176               processEntityTypeSelfLinks(typeLinksResult);
 177             } catch (Exception exc) {
 178               // TODO -> LOG, what should be logged here?
 179             }
 180
 181             return null;
 182           }
 183
 184         }, aaiExecutor).whenComplete((result, error) -> {
 185
 186           if (error != null) {
 187             LOG.error(AaiUiMsgs.ERROR_GENERIC,
 188                 "An error occurred getting data from AAI. Error = " + error.getMessage());
 189           }
 190         });
 191
 192       }
 193
 194       while (aaiWorkOnHand.get() != 0) {
 195
 196         if (LOG.isDebugEnabled()) {
 197           LOG.debug(AaiUiMsgs.WAIT_FOR_ALL_SELFLINKS_TO_BE_COLLECTED);
 198         }
 199
 200         Thread.sleep(1000);
 201       }
 202
 203       aaiWorkOnHand.set(selflinks.size());
 204       allWorkEnumerated = true;
 205       syncEntityTypes();
 206
 207       while (!isSyncDone()) {
 208         performRetrySync();
 209         Thread.sleep(1000);
 210       }
 211
 212       /*
 213        * Make sure we don't hang on to retries that failed which could cause issues during future
 214        * syncs
 215        */
 216       retryLimitTracker.clear();
 217
 218     } catch (Exception exc) {
 219       // TODO -> LOG, waht should be logged here?
 220     }
 221
 222     return OperationState.OK;
 223   }
 224
 225   /* (non-Javadoc)
 226    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync()
 227    */
 228   @Override
 229   public OperationState doSync() {
 230         this.syncDurationInMs = -1;
 231     String txnID = NodeUtils.getRandomTxnId();
 232     MdcContext.initialize(txnID, "SearchableEntitySynchronizer", "", "Sync", "");
 233
 234     resetCounters();
 235     this.allWorkEnumerated = false;
 236     syncStartedTimeStampInMs = System.currentTimeMillis();
 237     collectAllTheWork();
 238
 239     return OperationState.OK;
 240   }
 241
 242   /**
 243    * Process entity type self links.
 244    *
 245    * @param operationResult the operation result
 246    */
 247   private void processEntityTypeSelfLinks(OperationResult operationResult) {
 248
 249     JsonNode rootNode = null;
 250
 251     final String jsonResult = operationResult.getResult();
 252
 253     if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) {
 254
 255       try {
 256         rootNode = mapper.readTree(jsonResult);
 257       } catch (IOException exc) {
 258         String message =
 259             "Could not deserialize JSON (representing operation result) as node tree. " +
 260             "Operation result = " + jsonResult + ". " + exc.getLocalizedMessage();
 261         LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, message);
 262       }
 263
 264       JsonNode resultData = rootNode.get("result-data");
 265       ArrayNode resultDataArrayNode = null;
 266
 267       if (resultData.isArray()) {
 268         resultDataArrayNode = (ArrayNode) resultData;
 269
 270         Iterator<JsonNode> elementIterator = resultDataArrayNode.elements();
 271         JsonNode element = null;
 272
 273         while (elementIterator.hasNext()) {
 274           element = elementIterator.next();
 275
 276           final String resourceType = NodeUtils.getNodeFieldAsText(element, "resource-type");
 277           final String resourceLink = NodeUtils.getNodeFieldAsText(element, "resource-link");
 278
 279           OxmEntityDescriptor descriptor = null;
 280
 281           if (resourceType != null && resourceLink != null) {
 282
 283             descriptor = oxmModelLoader.getEntityDescriptor(resourceType);
 284
 285             if (descriptor == null) {
 286               LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, resourceType);
 287               // go to next element in iterator
 288               continue;
 289             }
 290
 291             if (descriptor.hasSearchableAttributes()) {
 292               selflinks.add(new SelfLinkDescriptor(resourceLink, SynchronizerConfiguration.NODES_ONLY_MODIFIER, resourceType));
 293             }
 294
 295           }
 296         }
 297       }
 298     }
 299
 300   }
 301
 302   /**
 303    * Sync entity types.
 304    */
 305   private void syncEntityTypes() {
 306
 307     while (selflinks.peek() != null) {
 308
 309       SelfLinkDescriptor linkDescriptor = selflinks.poll();
 310       aaiWorkOnHand.decrementAndGet();
 311
 312       OxmEntityDescriptor descriptor = null;
 313
 314       if (linkDescriptor.getSelfLink() != null && linkDescriptor.getEntityType() != null) {
 315
 316         descriptor = oxmModelLoader.getEntityDescriptor(linkDescriptor.getEntityType());
 317
 318         if (descriptor == null) {
 319           LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, linkDescriptor.getEntityType());
 320           // go to next element in iterator
 321           continue;
 322         }
 323
 324         NetworkTransaction txn = new NetworkTransaction();
 325         txn.setDescriptor(descriptor);
 326         txn.setLink(linkDescriptor.getSelfLink());
 327         txn.setOperationType(HttpMethod.GET);
 328         txn.setEntityType(linkDescriptor.getEntityType());
 329
 330         aaiWorkOnHand.incrementAndGet();
 331
 332         supplyAsync(new PerformActiveInventoryRetrieval(txn, aaiDataProvider), aaiExecutor)
 333             .whenComplete((result, error) -> {
 334
 335               aaiWorkOnHand.decrementAndGet();
 336
 337               if (error != null) {
 338                 LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_GENERIC, error.getLocalizedMessage());
 339               } else {
 340                 if (result == null) {
 341                   LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_FOR_SELF_LINK,
 342                       linkDescriptor.getSelfLink());
 343                 } else {
 344                   updateActiveInventoryCounters(result);
 345                   fetchDocumentForUpsert(result);
 346                 }
 347               }
 348             });
 349       }
 350
 351     }
 352
 353   }
 354
 355   /**
 356    * Perform document upsert.
 357    *
 358    * @param esGetTxn the es get txn
 359    * @param se the se
 360    */
 361   protected void performDocumentUpsert(NetworkTransaction esGetTxn, SearchableEntity se) {
 362     /**
 363      * <p>
 364      * <ul>
 365      * As part of the response processing we need to do the following:
 366      * <li>1. Extract the version (if present), it will be the ETAG when we use the
 367      * Search-Abstraction-Service
 368      * <li>2. Spawn next task which is to do the PUT operation into elastic with or with the version
 369      * tag
 370      * <li>a) if version is null or RC=404, then standard put, no _update with version tag
 371      * <li>b) if version != null, do PUT with _update?version= versionNumber in the URI to elastic
 372      * </ul>
 373      * </p>
 374      */
 375     String link = null;
 376     try {
 377       link = getElasticFullUrl("/" + se.getId(), getIndexName());
 378     } catch (Exception exc) {
 379       LOG.error(AaiUiMsgs.ES_LINK_UPSERT, exc.getLocalizedMessage());
 380       return;
 381     }
 382
 383     String versionNumber = null;
 384     boolean wasEntryDiscovered = false;
 385     if (esGetTxn.getOperationResult().getResultCode() == 404) {
 386       LOG.info(AaiUiMsgs.ES_SIMPLE_PUT, se.getEntityPrimaryKeyValue());
 387     } else if (esGetTxn.getOperationResult().getResultCode() == 200) {
 388       wasEntryDiscovered = true;
 389       try {
 390         versionNumber = NodeUtils.extractFieldValueFromObject(
 391             NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()),
 392             "_version");
 393       } catch (IOException exc) {
 394         String message =
 395             "Error extracting version number from response, aborting searchable entity sync of "
 396                 + se.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage();
 397         LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message);
 398         return;
 399       }
 400     } else {
 401       /*
 402        * Not being a 200 does not mean a failure. eg 201 is returned for created. TODO -> Should we
 403        * return.
 404        */
 405       LOG.error(AaiUiMsgs.ES_OPERATION_RETURN_CODE,
 406           String.valueOf(esGetTxn.getOperationResult().getResultCode()));
 407       return;
 408     }
 409
 410     try {
 411       String jsonPayload = null;
 412       if (wasEntryDiscovered) {
 413         try {
 414           ArrayList<JsonNode> sourceObject = new ArrayList<JsonNode>();
 415           NodeUtils.extractObjectsByKey(
 416               NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()),
 417               "_source", sourceObject);
 418
 419           if (!sourceObject.isEmpty()) {
 420             String responseSource = NodeUtils.convertObjectToJson(sourceObject.get(0), false);
 421             MergableEntity me = mapper.readValue(responseSource, MergableEntity.class);
 422             ObjectReader updater = mapper.readerForUpdating(me);
 423             MergableEntity merged = updater.readValue(se.getIndexDocumentJson());
 424             jsonPayload = mapper.writeValueAsString(merged);
 425           }
 426         } catch (IOException exc) {
 427           String message =
 428               "Error extracting source value from response, aborting searchable entity sync of "
 429                   + se.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage();
 430           LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message);
 431           return;
 432         }
 433       } else {
 434         jsonPayload = se.getIndexDocumentJson();
 435       }
 436
 437       if (wasEntryDiscovered) {
 438         if (versionNumber != null && jsonPayload != null) {
 439
 440           String requestPayload = esDataProvider.buildBulkImportOperationRequest(getIndexName(),
 441               ElasticSearchConfig.getConfig().getType(), se.getId(), versionNumber, jsonPayload);
 442
 443           NetworkTransaction transactionTracker = new NetworkTransaction();
 444           transactionTracker.setEntityType(esGetTxn.getEntityType());
 445           transactionTracker.setDescriptor(esGetTxn.getDescriptor());
 446           transactionTracker.setOperationType(HttpMethod.PUT);
 447
 448           esWorkOnHand.incrementAndGet();
 449           supplyAsync(new PerformElasticSearchUpdate(ElasticSearchConfig.getConfig().getBulkUrl(),
 450               requestPayload, esDataProvider, transactionTracker), esPutExecutor)
 451                   .whenComplete((result, error) -> {
 452
 453                     esWorkOnHand.decrementAndGet();
 454
 455                     if (error != null) {
 456                       String message = "Searchable entity sync UPDATE PUT error - "
 457                           + error.getLocalizedMessage();
 458                       LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 459                     } else {
 460                       updateElasticSearchCounters(result);
 461                       processStoreDocumentResult(result, esGetTxn, se);
 462                     }
 463                   });
 464         }
 465
 466       } else {
 467         if (link != null && jsonPayload != null) {
 468
 469           NetworkTransaction updateElasticTxn = new NetworkTransaction();
 470           updateElasticTxn.setLink(link);
 471           updateElasticTxn.setEntityType(esGetTxn.getEntityType());
 472           updateElasticTxn.setDescriptor(esGetTxn.getDescriptor());
 473           updateElasticTxn.setOperationType(HttpMethod.PUT);
 474
 475           esWorkOnHand.incrementAndGet();
 476           supplyAsync(new PerformElasticSearchPut(jsonPayload, updateElasticTxn, esDataProvider),
 477               esPutExecutor).whenComplete((result, error) -> {
 478
 479                 esWorkOnHand.decrementAndGet();
 480
 481                 if (error != null) {
 482                   String message =
 483                       "Searchable entity sync UPDATE PUT error - " + error.getLocalizedMessage();
 484                   LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 485                 } else {
 486                   updateElasticSearchCounters(result);
 487                   processStoreDocumentResult(result, esGetTxn, se);
 488                 }
 489               });
 490         }
 491       }
 492     } catch (Exception exc) {
 493       String message = "Exception caught during searchable entity sync PUT operation. Message - "
 494           + exc.getLocalizedMessage();
 495       LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 496     }
 497   }
 498
 499   /**
 500    * Populate searchable entity document.
 501    *
 502    * @param doc the doc
 503    * @param result the result
 504    * @param resultDescriptor the result descriptor
 505    * @throws JsonProcessingException the json processing exception
 506    * @throws IOException Signals that an I/O exception has occurred.
 507    */
 508   protected void populateSearchableEntityDocument(SearchableEntity doc, String result,
 509       OxmEntityDescriptor resultDescriptor) throws JsonProcessingException, IOException {
 510
 511     doc.setEntityType(resultDescriptor.getEntityName());
 512
 513     JsonNode entityNode = mapper.readTree(result);
 514
 515     List<String> primaryKeyValues = new ArrayList<String>();
 516     String pkeyValue = null;
 517
 518     for (String keyName : resultDescriptor.getPrimaryKeyAttributeName()) {
 519       pkeyValue = NodeUtils.getNodeFieldAsText(entityNode, keyName);
 520       if (pkeyValue != null) {
 521         primaryKeyValues.add(pkeyValue);
 522       } else {
 523         String message = "populateSearchableEntityDocument(), pKeyValue is null for entityType = "
 524             + resultDescriptor.getEntityName();
 525         LOG.warn(AaiUiMsgs.WARN_GENERIC, message);
 526       }
 527     }
 528
 529     final String primaryCompositeKeyValue = NodeUtils.concatArray(primaryKeyValues, "/");
 530     doc.setEntityPrimaryKeyValue(primaryCompositeKeyValue);
 531
 532     final List<String> searchTagFields = resultDescriptor.getSearchableAttributes();
 533
 534     /*
 535      * Based on configuration, use the configured field names for this entity-Type to build a
 536      * multi-value collection of search tags for elastic search entity search criteria.
 537      */
 538     for (String searchTagField : searchTagFields) {
 539       String searchTagValue = NodeUtils.getNodeFieldAsText(entityNode, searchTagField);
 540       if (searchTagValue != null && !searchTagValue.isEmpty()) {
 541         doc.addSearchTagWithKey(searchTagValue, searchTagField);
 542       }
 543     }
 544   }
 545
 546   /**
 547    * Fetch document for upsert.
 548    *
 549    * @param txn the txn
 550    */
 551   private void fetchDocumentForUpsert(NetworkTransaction txn) {
 552     if (!txn.getOperationResult().wasSuccessful()) {
 553       String message = "Self link failure. Result - " + txn.getOperationResult().getResult();
 554       LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
 555       return;
 556     }
 557
 558     try {
 559       if (txn.getDescriptor().hasSearchableAttributes()) {
 560
 561         final String jsonResult = txn.getOperationResult().getResult();
 562         if (jsonResult != null && jsonResult.length() > 0) {
 563
 564           SearchableEntity se = new SearchableEntity(oxmModelLoader);
 565           se.setLink(ActiveInventoryConfig.extractResourcePath(txn.getLink()));
 566           populateSearchableEntityDocument(se, jsonResult, txn.getDescriptor());
 567           se.deriveFields();
 568
 569           String link = null;
 570           try {
 571             link = getElasticFullUrl("/" + se.getId(), getIndexName());
 572           } catch (Exception exc) {
 573             LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_QUERY, exc.getLocalizedMessage());
 574           }
 575
 576           if (link != null) {
 577             NetworkTransaction n2 = new NetworkTransaction();
 578             n2.setLink(link);
 579             n2.setEntityType(txn.getEntityType());
 580             n2.setDescriptor(txn.getDescriptor());
 581             n2.setOperationType(HttpMethod.GET);
 582
 583             esWorkOnHand.incrementAndGet();
 584
 585             supplyAsync(new PerformElasticSearchRetrieval(n2, esDataProvider), esExecutor)
 586                 .whenComplete((result, error) -> {
 587
 588                   esWorkOnHand.decrementAndGet();
 589
 590                   if (error != null) {
 591                     LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED, error.getLocalizedMessage());
 592                   } else {
 593                     updateElasticSearchCounters(result);
 594                     performDocumentUpsert(result, se);
 595                   }
 596                 });
 597           }
 598         }
 599
 600       }
 601     } catch (JsonProcessingException exc) {
 602       // TODO -> LOG, waht should be logged here?
 603     } catch (IOException exc) {
 604       // TODO -> LOG, waht should be logged here?
 605     }
 606   }
 607
 608   /**
 609    * Process store document result.
 610    *
 611    * @param esPutResult the es put result
 612    * @param esGetResult the es get result
 613    * @param se the se
 614    */
 615   private void processStoreDocumentResult(NetworkTransaction esPutResult,
 616       NetworkTransaction esGetResult, SearchableEntity se) {
 617
 618     OperationResult or = esPutResult.getOperationResult();
 619
 620     if (!or.wasSuccessful()) {
 621       if (or.getResultCode() == VERSION_CONFLICT_EXCEPTION_CODE) {
 622
 623         if (shouldAllowRetry(se.getId())) {
 624           esWorkOnHand.incrementAndGet();
 625
 626           RetrySearchableEntitySyncContainer rsc =
 627               new RetrySearchableEntitySyncContainer(esGetResult, se);
 628           retryQueue.push(rsc);
 629
 630           String message = "Store document failed during searchable entity synchronization"
 631               + " due to version conflict. Entity will be re-synced.";
 632           LOG.warn(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 633         }
 634       } else {
 635         String message =
 636             "Store document failed during searchable entity synchronization with result code "
 637                 + or.getResultCode() + " and result message " + or.getResult();
 638         LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 639       }
 640     }
 641   }
 642
 643   /**
 644    * Perform retry sync.
 645    */
 646   private void performRetrySync() {
 647     while (retryQueue.peek() != null) {
 648
 649       RetrySearchableEntitySyncContainer rsc = retryQueue.poll();
 650       if (rsc != null) {
 651
 652         SearchableEntity se = rsc.getSearchableEntity();
 653         NetworkTransaction txn = rsc.getNetworkTransaction();
 654
 655         String link = null;
 656         try {
 657           /*
 658            * In this retry flow the se object has already derived its fields
 659            */
 660           link = getElasticFullUrl("/" + se.getId(), getIndexName());
 661         } catch (Exception exc) {
 662           LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_URI, exc.getLocalizedMessage());
 663         }
 664
 665         if (link != null) {
 666           NetworkTransaction retryTransaction = new NetworkTransaction();
 667           retryTransaction.setLink(link);
 668           retryTransaction.setEntityType(txn.getEntityType());
 669           retryTransaction.setDescriptor(txn.getDescriptor());
 670           retryTransaction.setOperationType(HttpMethod.GET);
 671
 672           /*
 673            * IMPORTANT - DO NOT incrementAndGet the esWorkOnHand as this is a retry flow! We already
 674            * called incrementAndGet when queuing the failed PUT!
 675            */
 676
 677           supplyAsync(new PerformElasticSearchRetrieval(retryTransaction, esDataProvider),
 678               esExecutor).whenComplete((result, error) -> {
 679
 680                 esWorkOnHand.decrementAndGet();
 681
 682                 if (error != null) {
 683                   LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED_RESYNC, error.getLocalizedMessage());
 684                 } else {
 685                   updateElasticSearchCounters(result);
 686                   performDocumentUpsert(result, se);
 687                 }
 688               });
 689         }
 690
 691       }
 692     }
 693   }
 694
 695   /**
 696    * Should allow retry.
 697    *
 698    * @param id the id
 699    * @return true, if successful
 700    */
 701   private boolean shouldAllowRetry(String id) {
 702     boolean isRetryAllowed = true;
 703     if (retryLimitTracker.get(id) != null) {
 704       Integer currentCount = retryLimitTracker.get(id);
 705       if (currentCount.intValue() >= RETRY_COUNT_PER_ENTITY_LIMIT.intValue()) {
 706         isRetryAllowed = false;
 707         String message = "Searchable entity re-sync limit reached for " + id
 708             + ", re-sync will no longer be attempted for this entity";
 709         LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 710       } else {
 711         Integer newCount = new Integer(currentCount.intValue() + 1);
 712         retryLimitTracker.put(id, newCount);
 713       }
 714     } else {
 715       Integer firstRetryCount = new Integer(1);
 716       retryLimitTracker.put(id, firstRetryCount);
 717     }
 718
 719     return isRetryAllowed;
 720   }
 721
 722   @Override
 723   public SynchronizerState getState() {
 724     if (!isSyncDone()) {
 725       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 726     }
 727
 728     return SynchronizerState.IDLE;
 729
 730   }
 731
 732   /* (non-Javadoc)
 733    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean)
 734    */
 735   @Override
 736   public String getStatReport(boolean showFinalReport) {
 737           syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 738           return this.getStatReport(syncDurationInMs, showFinalReport);
 739   }
 740
 741   /* (non-Javadoc)
 742    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown()
 743    */
 744   @Override
 745   public void shutdown() {
 746     this.shutdownExecutors();
 747   }
 748
 749   @Override
 750   protected boolean isSyncDone() {
 751     int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get();
 752
 753     if (totalWorkOnHand > 0 || !allWorkEnumerated) {
 754       return false;
 755     }
 756
 757     return true;
 758   }
 759
 760 }