src/main/java/org/onap/aai/sparky/aggregation/sync/HistoricalEntitySummarizer.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.aggregation.sync;
  24
  25 import static java.util.concurrent.CompletableFuture.supplyAsync;
  26
  27 import java.io.IOException;
  28 import java.sql.Timestamp;
  29 import java.text.SimpleDateFormat;
  30 import java.util.Collection;
  31 import java.util.EnumSet;
  32 import java.util.Map;
  33 import java.util.Map.Entry;
  34 import java.util.Set;
  35 import java.util.concurrent.ConcurrentHashMap;
  36 import java.util.concurrent.atomic.AtomicInteger;
  37 import java.util.function.Supplier;
  38
  39 import javax.json.Json;
  40 import javax.ws.rs.core.MediaType;
  41
  42 import org.onap.aai.cl.api.Logger;
  43 import org.onap.aai.cl.eelf.LoggerFactory;
  44 import org.onap.aai.cl.mdc.MdcContext;
  45 import org.onap.aai.restclient.client.OperationResult;
  46 import org.onap.aai.sparky.config.oxm.SearchableEntityLookup;
  47 import org.onap.aai.sparky.config.oxm.SearchableOxmEntityDescriptor;
  48 import org.onap.aai.sparky.dal.rest.HttpMethod;
  49 import org.onap.aai.sparky.logging.AaiUiMsgs;
  50 import org.onap.aai.sparky.sync.AbstractEntitySynchronizer;
  51 import org.onap.aai.sparky.sync.IndexSynchronizer;
  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.enumeration.OperationState;
  55 import org.onap.aai.sparky.sync.enumeration.SynchronizerState;
  56 import org.onap.aai.sparky.util.NodeUtils;
  57 import org.slf4j.MDC;
  58
  59 import com.fasterxml.jackson.databind.JsonNode;
  60 import com.fasterxml.jackson.databind.node.ArrayNode;
  61
  62 /**
  63  * The Class HistoricalEntitySummarizer.
  64  */
  65 public class HistoricalEntitySummarizer extends AbstractEntitySynchronizer
  66     implements IndexSynchronizer {
  67
  68   private static final Logger LOG =
  69       LoggerFactory.getInstance().getLogger(HistoricalEntitySummarizer.class);
  70   private static final String INSERTION_DATE_TIME_FORMAT = "yyyyMMdd'T'HHmmssZ";
  71
  72   private boolean allWorkEnumerated;
  73   private ConcurrentHashMap<String, AtomicInteger> entityCounters;
  74   private boolean syncInProgress;
  75   private Map<String, String> contextMap;
  76   private ElasticSearchSchemaConfig schemaConfig;
  77
  78   /**
  79    * Instantiates a new historical entity summarizer.
  80    *
  81    * @param indexName the index name
  82    * @throws Exception the exception
  83    */
  84   public HistoricalEntitySummarizer(ElasticSearchSchemaConfig schemaConfig, int internalSyncWorkers,
  85       int aaiWorkers, int esWorkers, NetworkStatisticsConfig aaiStatConfig,
  86       NetworkStatisticsConfig esStatConfig) throws Exception {
  87     super(LOG, "HES", internalSyncWorkers, aaiWorkers, esWorkers, schemaConfig.getIndexName(),
  88         aaiStatConfig, esStatConfig);
  89
  90     this.schemaConfig = schemaConfig;
  91     this.allWorkEnumerated = false;
  92     this.entityCounters = new ConcurrentHashMap<String, AtomicInteger>();
  93     this.synchronizerName = "Historical Entity Summarizer";
  94     this.enabledStatFlags = EnumSet.of(StatFlag.AAI_REST_STATS, StatFlag.ES_REST_STATS);
  95     this.syncInProgress = false;
  96     this.contextMap = MDC.getCopyOfContextMap();
  97     this.syncDurationInMs = -1;
  98   }
  99
 100   /**
 101    * Collect all the work.
 102    *
 103    * @return the operation state
 104    */
 105   private OperationState collectAllTheWork() {
 106
 107     Map<String, SearchableOxmEntityDescriptor> descriptorMap =
 108         SearchableEntityLookup.getInstance().getSearchableEntityDescriptors();
 109
 110     if (descriptorMap.isEmpty()) {
 111       LOG.error(AaiUiMsgs.OXM_FAILED_RETRIEVAL, "historical entities");
 112
 113       return OperationState.ERROR;
 114     }
 115
 116     Collection<String> entityTypes = descriptorMap.keySet();
 117
 118     AtomicInteger asyncWoH = new AtomicInteger(0);
 119
 120     asyncWoH.set(entityTypes.size());
 121
 122     try {
 123       for (String entityType : entityTypes) {
 124
 125         supplyAsync(new Supplier<Void>() {
 126
 127           @Override
 128           public Void get() {
 129             MDC.setContextMap(contextMap);
 130             try {
 131               OperationResult typeLinksResult = aaiAdapter.getSelfLinksByEntityType(entityType);
 132               updateActiveInventoryCounters(HttpMethod.GET, entityType, typeLinksResult);
 133               processEntityTypeSelfLinks(entityType, typeLinksResult);
 134             } catch (Exception exc) {
 135               LOG.error(AaiUiMsgs.ERROR_GETTING_DATA_FROM_AAI, exc.getMessage());
 136
 137             }
 138
 139             return null;
 140           }
 141
 142         }, aaiExecutor).whenComplete((result, error) -> {
 143
 144           asyncWoH.decrementAndGet();
 145
 146           if (error != null) {
 147             LOG.error(AaiUiMsgs.HISTORICAL_COLLECT_ERROR, error.getMessage());
 148           }
 149
 150         });
 151
 152       }
 153
 154
 155       while (asyncWoH.get() > 0) {
 156
 157         if (LOG.isDebugEnabled()) {
 158           LOG.debug(AaiUiMsgs.DEBUG_GENERIC,
 159               indexName + " summarizer waiting for all the links to be processed.");
 160         }
 161
 162         Thread.sleep(250);
 163       }
 164
 165       esWorkOnHand.set(entityCounters.size());
 166
 167       // start doing the real work
 168       allWorkEnumerated = true;
 169
 170       insertEntityTypeCounters();
 171
 172       if (LOG.isDebugEnabled()) {
 173
 174         StringBuilder sb = new StringBuilder(128);
 175
 176         sb.append("\n\nHistorical Entity Counters:");
 177
 178         for (Entry<String, AtomicInteger> entry : entityCounters.entrySet()) {
 179           sb.append("\n").append(entry.getKey()).append(" = ").append(entry.getValue().get());
 180         }
 181
 182         LOG.debug(AaiUiMsgs.DEBUG_GENERIC, sb.toString());
 183
 184       }
 185
 186     } catch (Exception exc) {
 187       LOG.error(AaiUiMsgs.HISTORICAL_COLLECT_ERROR, exc.getMessage());
 188
 189
 190       esWorkOnHand.set(0);
 191       allWorkEnumerated = true;
 192
 193       return OperationState.ERROR;
 194     }
 195
 196     return OperationState.OK;
 197
 198   }
 199
 200   /*
 201    * (non-Javadoc)
 202    *
 203    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync()
 204    */
 205   @Override
 206   public OperationState doSync() {
 207     this.syncDurationInMs = -1;
 208     String txnID = NodeUtils.getRandomTxnId();
 209     MdcContext.initialize(txnID, "HistoricalEntitySynchronizer", "", "Sync", "");
 210
 211     if (syncInProgress) {
 212       LOG.info(AaiUiMsgs.HISTORICAL_SYNC_PENDING);
 213       return OperationState.PENDING;
 214     }
 215
 216     clearCache();
 217
 218     syncInProgress = true;
 219     this.syncStartedTimeStampInMs = System.currentTimeMillis();
 220     allWorkEnumerated = false;
 221
 222     return collectAllTheWork();
 223   }
 224
 225   /**
 226    * Process entity type self links.
 227    *
 228    * @param entityType the entity type
 229    * @param operationResult the operation result
 230    */
 231   private void processEntityTypeSelfLinks(String entityType, OperationResult operationResult) {
 232
 233     JsonNode rootNode = null;
 234
 235     final String jsonResult = operationResult.getResult();
 236
 237     if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) {
 238
 239       try {
 240         rootNode = mapper.readTree(jsonResult);
 241       } catch (IOException exc) {
 242         LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, exc.getMessage());
 243         return;
 244       }
 245
 246       JsonNode resultData = rootNode.get("result-data");
 247       ArrayNode resultDataArrayNode = null;
 248
 249       if (resultData != null && resultData.isArray()) {
 250         resultDataArrayNode = (ArrayNode) resultData;
 251         entityCounters.put(entityType, new AtomicInteger(resultDataArrayNode.size()));
 252       }
 253     }
 254
 255   }
 256
 257   /**
 258    * Insert entity type counters.
 259    */
 260   private void insertEntityTypeCounters() {
 261
 262     if (esWorkOnHand.get() <= 0) {
 263       return;
 264     }
 265
 266     SimpleDateFormat dateFormat = new SimpleDateFormat(INSERTION_DATE_TIME_FORMAT);
 267     Timestamp timestamp = new Timestamp(System.currentTimeMillis());
 268     String currentFormattedTimeStamp = dateFormat.format(timestamp);
 269
 270     Set<Entry<String, AtomicInteger>> entityCounterEntries = entityCounters.entrySet();
 271
 272     for (Entry<String, AtomicInteger> entityCounterEntry : entityCounterEntries) {
 273
 274       supplyAsync(new Supplier<Void>() {
 275
 276         @Override
 277         public Void get() {
 278           MDC.setContextMap(contextMap);
 279           String jsonString =
 280               Json.createObjectBuilder().add("count", entityCounterEntry.getValue().get())
 281                   .add("entityType", entityCounterEntry.getKey())
 282                   .add("timestamp", currentFormattedTimeStamp).build().toString();
 283
 284           String link = null;
 285           try {
 286             link = getElasticFullUrl("", indexName);
 287             OperationResult or =
 288                 elasticSearchAdapter.doPost(link, jsonString, MediaType.APPLICATION_JSON_TYPE);
 289             updateElasticSearchCounters(HttpMethod.POST, entityCounterEntry.getKey(), or);
 290           } catch (Exception exc) {
 291             LOG.error(AaiUiMsgs.ES_STORE_FAILURE, exc.getMessage());
 292           }
 293
 294           return null;
 295         }
 296
 297       }, esExecutor).whenComplete((result, error) -> {
 298
 299         esWorkOnHand.decrementAndGet();
 300
 301       });
 302
 303     }
 304
 305     while (esWorkOnHand.get() > 0) {
 306
 307       try {
 308         Thread.sleep(500);
 309       } catch (InterruptedException exc) {
 310         LOG.error(AaiUiMsgs.INTERRUPTED, "historical Entities", exc.getMessage());
 311       }
 312     }
 313
 314   }
 315
 316   @Override
 317   public SynchronizerState getState() {
 318
 319     if (!isSyncDone()) {
 320       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 321     }
 322
 323     return SynchronizerState.IDLE;
 324
 325   }
 326
 327   /*
 328    * (non-Javadoc)
 329    *
 330    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean)
 331    */
 332   @Override
 333   public String getStatReport(boolean showFinalReport) {
 334     syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 335     return this.getStatReport(syncDurationInMs, showFinalReport);
 336   }
 337
 338   /*
 339    * (non-Javadoc)
 340    *
 341    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown()
 342    */
 343   @Override
 344   public void shutdown() {
 345     this.shutdownExecutors();
 346   }
 347
 348   @Override
 349   protected boolean isSyncDone() {
 350
 351     int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get();
 352
 353     if (LOG.isDebugEnabled()) {
 354       LOG.debug(AaiUiMsgs.DEBUG_GENERIC, indexName + ", isSyncDone(), totalWorkOnHand = "
 355           + totalWorkOnHand + " all work enumerated = " + allWorkEnumerated);
 356     }
 357
 358     if (totalWorkOnHand > 0 || !allWorkEnumerated) {
 359       return false;
 360     }
 361
 362     this.syncInProgress = false;
 363
 364     return true;
 365   }
 366
 367   /*
 368    * (non-Javadoc)
 369    *
 370    * @see org.openecomp.sparky.synchronizer.AbstractEntitySynchronizer#clearCache()
 371    */
 372   @Override
 373   public void clearCache() {
 374
 375     if (syncInProgress) {
 376       LOG.debug(AaiUiMsgs.DEBUG_GENERIC,
 377           "Historical Entity Summarizer in progress, request to clear cache ignored");
 378       return;
 379     }
 380
 381     super.clearCache();
 382     this.resetCounters();
 383     if (entityCounters != null) {
 384       entityCounters.clear();
 385     }
 386
 387     allWorkEnumerated = false;
 388
 389   }
 390
 391 }