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