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