gerrit.onap Code Review - aai/sparky-be.git/blob

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