src/main/java/org/openecomp/sparky/synchronizer/GeoSynchronizer.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 java.io.IOException;
  28 import java.util.ArrayList;
  29 import java.util.Collection;
  30 import java.util.Deque;
  31 import java.util.Iterator;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.concurrent.ConcurrentLinkedDeque;
  35 import java.util.function.Supplier;
  36
  37 import org.openecomp.cl.api.Logger;
  38 import org.openecomp.cl.eelf.LoggerFactory;
  39 import org.openecomp.cl.mdc.MdcContext;
  40 import org.openecomp.sparky.config.oxm.OxmEntityDescriptor;
  41 import org.openecomp.sparky.dal.NetworkTransaction;
  42 import org.openecomp.sparky.dal.elasticsearch.config.ElasticSearchConfig;
  43 import org.openecomp.sparky.dal.rest.HttpMethod;
  44 import org.openecomp.sparky.dal.rest.OperationResult;
  45 import org.openecomp.sparky.inventory.entity.GeoIndexDocument;
  46 import org.openecomp.sparky.logging.AaiUiMsgs;
  47 import org.openecomp.sparky.synchronizer.entity.SelfLinkDescriptor;
  48 import org.openecomp.sparky.synchronizer.enumeration.OperationState;
  49 import org.openecomp.sparky.synchronizer.enumeration.SynchronizerState;
  50 import org.openecomp.sparky.synchronizer.task.PerformActiveInventoryRetrieval;
  51 import org.openecomp.sparky.synchronizer.task.StoreDocumentTask;
  52 import org.openecomp.sparky.util.NodeUtils;
  53 import org.slf4j.MDC;
  54
  55 import com.fasterxml.jackson.core.JsonProcessingException;
  56 import com.fasterxml.jackson.databind.JsonNode;
  57 import com.fasterxml.jackson.databind.node.ArrayNode;
  58
  59
  60 /**
  61  * The Class GeoSynchronizer.
  62  */
  63 public class GeoSynchronizer extends AbstractEntitySynchronizer implements IndexSynchronizer {
  64
  65   private static final Logger LOG = LoggerFactory.getInstance().getLogger(GeoSynchronizer.class);
  66
  67   private boolean allWorkEnumerated;
  68   private Deque<SelfLinkDescriptor> selflinks;
  69
  70   private ElasticSearchConfig elasticConfig = null;
  71   private Map<String, OxmEntityDescriptor> geoDescriptorMap = null;
  72
  73   /**
  74    * Instantiates a new geo synchronizer.
  75    *
  76    * @param indexName the index name
  77    * @throws Exception the exception
  78    */
  79   public GeoSynchronizer(String indexName) throws Exception {
  80
  81     super(LOG, "GEO", 2, 5, 5, indexName);
  82     this.allWorkEnumerated = false;
  83     this.selflinks = new ConcurrentLinkedDeque<SelfLinkDescriptor>();
  84     this.synchronizerName = "Geo Synchronizer";
  85     this.geoDescriptorMap = oxmModelLoader.getGeoEntityDescriptors();
  86     this.aaiEntityStats.initializeCountersFromOxmEntityDescriptors(geoDescriptorMap);
  87     this.esEntityStats.initializeCountersFromOxmEntityDescriptors(geoDescriptorMap);
  88     this.syncDurationInMs = -1;
  89   }
  90
  91
  92   /* (non-Javadoc)
  93    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync()
  94    */
  95   @Override
  96   public OperationState doSync() {
  97     resetCounters();
  98     allWorkEnumerated = false;
  99     syncStartedTimeStampInMs = System.currentTimeMillis();
 100     String txnID = NodeUtils.getRandomTxnId();
 101     MdcContext.initialize(txnID, "GeoSynchronizer", "", "Sync", "");
 102
 103     collectAllTheWork();
 104     return OperationState.OK;
 105   }
 106
 107
 108   /**
 109    * Collect all the work.
 110    *
 111    * @return the operation state
 112    */
 113   public OperationState collectAllTheWork() {
 114         final Map<String,String> contextMap = MDC.getCopyOfContextMap();
 115     if (elasticConfig == null) {
 116       try {
 117         elasticConfig = ElasticSearchConfig.getConfig();
 118       } catch (Exception exc) {
 119         LOG.error(AaiUiMsgs.CONFIGURATION_ERROR, "Search");
 120       }
 121     }
 122
 123     if (geoDescriptorMap.isEmpty()) {
 124       setShouldSkipSync(true);
 125       LOG.error(AaiUiMsgs.OXM_FAILED_RETRIEVAL, "geo entities");
 126       return OperationState.ERROR;
 127     }
 128
 129     Collection<String> syncTypes = geoDescriptorMap.keySet();
 130
 131     try {
 132
 133       /*
 134        * launch a parallel async thread to process the documents for each entity-type (to max the of
 135        * the configured executor anyway)
 136        */
 137
 138       aaiWorkOnHand.set(syncTypes.size());
 139
 140       for (String key : syncTypes) {
 141
 142         supplyAsync(new Supplier<Void>() {
 143
 144           @Override
 145           public Void get() {
 146                 MDC.setContextMap(contextMap);
 147             OperationResult typeLinksResult = null;
 148             try {
 149               typeLinksResult = aaiDataProvider.getSelfLinksByEntityType(key);
 150               aaiWorkOnHand.decrementAndGet();
 151               processEntityTypeSelfLinks(typeLinksResult);
 152             } catch (Exception exc) {
 153               LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, exc);
 154             }
 155
 156             return null;
 157           }
 158
 159         }, aaiExecutor).whenComplete((result, error) -> {
 160
 161           if (error != null) {
 162             LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, error.getMessage());
 163           }
 164         });
 165
 166       }
 167
 168       while (aaiWorkOnHand.get() != 0) {
 169
 170         if (LOG.isDebugEnabled()) {
 171           LOG.debug(AaiUiMsgs.WAIT_FOR_ALL_SELFLINKS_TO_BE_COLLECTED);
 172         }
 173
 174         Thread.sleep(1000);
 175       }
 176
 177       aaiWorkOnHand.set(selflinks.size());
 178       allWorkEnumerated = true;
 179       syncEntityTypes();
 180
 181     } catch (Exception exc) {
 182       LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, exc);
 183     }
 184     return OperationState.OK;
 185   }
 186
 187   /**
 188    * Sync entity types.
 189    */
 190   private void syncEntityTypes() {
 191
 192     while (selflinks.peek() != null) {
 193
 194       SelfLinkDescriptor linkDescriptor = selflinks.poll();
 195       aaiWorkOnHand.decrementAndGet();
 196
 197       OxmEntityDescriptor descriptor = null;
 198
 199       if (linkDescriptor.getSelfLink() != null && linkDescriptor.getEntityType() != null) {
 200
 201         descriptor = oxmModelLoader.getEntityDescriptor(linkDescriptor.getEntityType());
 202
 203         if (descriptor == null) {
 204           LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, linkDescriptor.getEntityType());
 205           // go to next element in iterator
 206           continue;
 207         }
 208
 209         NetworkTransaction txn = new NetworkTransaction();
 210         txn.setDescriptor(descriptor);
 211         txn.setLink(linkDescriptor.getSelfLink());
 212         txn.setOperationType(HttpMethod.GET);
 213         txn.setEntityType(linkDescriptor.getEntityType());
 214
 215         aaiWorkOnHand.incrementAndGet();
 216
 217         supplyAsync(new PerformActiveInventoryRetrieval(txn, aaiDataProvider), aaiExecutor)
 218             .whenComplete((result, error) -> {
 219
 220               aaiWorkOnHand.decrementAndGet();
 221
 222               if (error != null) {
 223                 LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, error.getMessage());
 224               } else {
 225                 if (result == null) {
 226                   LOG.error(AaiUiMsgs.SELF_LINK_GET_NO_RESPONSE, linkDescriptor.getSelfLink());
 227                 } else {
 228                   processEntityTypeSelfLinkResult(result);
 229                 }
 230               }
 231             });
 232       }
 233     }
 234   }
 235
 236   /**
 237    * Process entity type self links.
 238    *
 239    * @param operationResult the operation result
 240    */
 241   private void processEntityTypeSelfLinks(OperationResult operationResult) {
 242
 243     JsonNode rootNode = null;
 244
 245     final String jsonResult = operationResult.getResult();
 246
 247     if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) {
 248
 249       try {
 250         rootNode = mapper.readTree(jsonResult);
 251       } catch (IOException exc) {
 252         LOG.error(AaiUiMsgs.ERROR_GENERIC, exc);
 253       }
 254
 255       JsonNode resultData = rootNode.get("result-data");
 256       ArrayNode resultDataArrayNode = null;
 257
 258       if (resultData.isArray()) {
 259         resultDataArrayNode = (ArrayNode) resultData;
 260
 261         Iterator<JsonNode> elementIterator = resultDataArrayNode.elements();
 262         JsonNode element = null;
 263
 264         while (elementIterator.hasNext()) {
 265           element = elementIterator.next();
 266
 267           final String resourceType = NodeUtils.getNodeFieldAsText(element, "resource-type");
 268           final String resourceLink = NodeUtils.getNodeFieldAsText(element, "resource-link");
 269
 270           if (resourceType != null && resourceLink != null) {
 271
 272             if (geoDescriptorMap.containsKey(resourceType)) {
 273               selflinks.add(new SelfLinkDescriptor(resourceLink + "?nodes-only", resourceType));
 274             } else {
 275               LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, resourceType);
 276               // go to next element in iterator
 277               continue;
 278             }
 279
 280           }
 281         }
 282       }
 283     }
 284
 285   }
 286
 287   /**
 288    * Process entity type self link result.
 289    *
 290    * @param txn the txn
 291    */
 292   private void processEntityTypeSelfLinkResult(NetworkTransaction txn) {
 293
 294     updateActiveInventoryCounters(txn);
 295
 296     if (!txn.getOperationResult().wasSuccessful()) {
 297       return;
 298     }
 299
 300     try {
 301       if (!(txn.getDescriptor().getGeoLatName().isEmpty()
 302           && txn.getDescriptor().getGeoLongName().isEmpty())) {
 303
 304         GeoIndexDocument geoDoc = new GeoIndexDocument(oxmModelLoader);
 305
 306         final String jsonResult = txn.getOperationResult().getResult();
 307
 308         if (jsonResult != null && jsonResult.length() > 0) {
 309
 310           populateGeoDocument(geoDoc, jsonResult, txn.getDescriptor(), txn.getLink());
 311
 312           if (!geoDoc.isValidGeoDocument()) {
 313
 314             LOG.info(AaiUiMsgs.GEO_SYNC_IGNORING_ENTITY, geoDoc.getEntityType(), geoDoc.toString());
 315
 316           } else {
 317
 318             String link = null;
 319             try {
 320               link = getElasticFullUrl("/" + geoDoc.getId(), getIndexName(), "default");
 321             } catch (Exception exc) {
 322               LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_URI, exc);
 323             }
 324
 325             if (link != null) {
 326
 327               NetworkTransaction n2 = new NetworkTransaction();
 328               n2.setLink(link);
 329               n2.setEntityType(txn.getEntityType());
 330               n2.setDescriptor(txn.getDescriptor());
 331               n2.setOperationType(HttpMethod.PUT);
 332
 333               esWorkOnHand.incrementAndGet();
 334
 335               supplyAsync(new StoreDocumentTask(geoDoc, n2, esDataProvider), esExecutor)
 336                   .whenComplete((result, error) -> {
 337
 338                     esWorkOnHand.decrementAndGet();
 339
 340                     if (error != null) {
 341                       LOG.error(AaiUiMsgs.ES_STORE_FAILURE, error.getMessage());
 342                     } else {
 343                       updateElasticSearchCounters(result);
 344                       processStoreDocumentResult(result);
 345                     }
 346                   });
 347             }
 348           }
 349         }
 350       }
 351     } catch (JsonProcessingException exc) {
 352       LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, exc);
 353     } catch (IOException exc) {
 354       LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, exc);
 355     }
 356
 357     return;
 358   }
 359
 360
 361   /**
 362    * Process store document result.
 363    *
 364    * @param txn the txn
 365    */
 366   private void processStoreDocumentResult(NetworkTransaction txn) {
 367
 368     OperationResult or = txn.getOperationResult();
 369
 370     if (!or.wasSuccessful()) {
 371       LOG.error(AaiUiMsgs.ES_STORE_FAILURE, or.toString());
 372       /*
 373        * if(or.getResultCode() != 404 || (or.getResultCode() == 404 &&
 374        * !synchronizerConfig.isResourceNotFoundErrorsSupressed())) { logger.error(
 375        * "Skipping failed resource = " + "link" + " RC=[" + or.getResultCode() + "]. Message: " +
 376        * or.getResult()); }
 377        */
 378
 379     }
 380
 381   }
 382
 383
 384   @Override
 385   public SynchronizerState getState() {
 386
 387     if (!isSyncDone()) {
 388       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 389     }
 390
 391     return SynchronizerState.IDLE;
 392
 393   }
 394
 395   /* (non-Javadoc)
 396    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean)
 397    */
 398   @Override
 399   public String getStatReport(boolean showFinalReport) {
 400           syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 401           return this.getStatReport(syncDurationInMs, showFinalReport);
 402   }
 403
 404   /* (non-Javadoc)
 405    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown()
 406    */
 407   @Override
 408   public void shutdown() {
 409     this.shutdownExecutors();
 410   }
 411
 412   /**
 413    * Populate geo document.
 414    *
 415    * @param doc the doc
 416    * @param result the result
 417    * @param resultDescriptor the result descriptor
 418    * @param entityLink the entity link
 419    * @throws JsonProcessingException the json processing exception
 420    * @throws IOException Signals that an I/O exception has occurred.
 421    */
 422   protected void populateGeoDocument(GeoIndexDocument doc, String result,
 423       OxmEntityDescriptor resultDescriptor, String entityLink)
 424           throws JsonProcessingException, IOException {
 425
 426     doc.setSelfLink(entityLink);
 427     doc.setEntityType(resultDescriptor.getEntityName());
 428
 429     JsonNode entityNode = mapper.readTree(result);
 430
 431     List<String> primaryKeyValues = new ArrayList<String>();
 432     String pkeyValue = null;
 433
 434     for (String keyName : resultDescriptor.getPrimaryKeyAttributeName()) {
 435       pkeyValue = NodeUtils.getNodeFieldAsText(entityNode, keyName);
 436       if (pkeyValue != null) {
 437         primaryKeyValues.add(pkeyValue);
 438       } else {
 439         LOG.warn(AaiUiMsgs.ES_PKEYVALUE_NULL, resultDescriptor.getEntityName());
 440       }
 441     }
 442
 443     final String primaryCompositeKeyValue = NodeUtils.concatArray(primaryKeyValues, "/");
 444     doc.setEntityPrimaryKeyValue(primaryCompositeKeyValue);
 445     String geoLatKey = resultDescriptor.getGeoLatName();
 446     String geoLongKey = resultDescriptor.getGeoLongName();
 447
 448     doc.setLatitude(NodeUtils.getNodeFieldAsText(entityNode, geoLatKey));
 449     doc.setLongitude(NodeUtils.getNodeFieldAsText(entityNode, geoLongKey));
 450     doc.deriveFields();
 451
 452   }
 453
 454   @Override
 455   protected boolean isSyncDone() {
 456     int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get();
 457
 458     if (totalWorkOnHand > 0 || !allWorkEnumerated) {
 459       return false;
 460     }
 461
 462     return true;
 463   }
 464
 465 }