catalog-model/src/main/java/org/openecomp/sdc/be/model/cache/ApplicationDataTypeCache.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.openecomp.sdc.be.model.cache;
  21
  22 import fj.data.Either;
  23 import java.util.ArrayList;
  24 import java.util.Collections;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.Optional;
  29 import java.util.concurrent.Executors;
  30 import java.util.concurrent.ScheduledExecutorService;
  31 import java.util.concurrent.ScheduledFuture;
  32 import java.util.concurrent.TimeUnit;
  33 import java.util.concurrent.locks.ReentrantReadWriteLock;
  34 import javax.annotation.PostConstruct;
  35 import javax.annotation.PreDestroy;
  36 import lombok.AccessLevel;
  37 import lombok.Getter;
  38 import org.apache.commons.collections.CollectionUtils;
  39 import org.apache.commons.collections.MapUtils;
  40 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
  41 import org.openecomp.sdc.be.config.BeEcompErrorManager;
  42 import org.openecomp.sdc.be.config.BeEcompErrorManager.ErrorSeverity;
  43 import org.openecomp.sdc.be.config.Configuration.ApplicationL1CacheInfo;
  44 import org.openecomp.sdc.be.config.ConfigurationManager;
  45 import org.openecomp.sdc.be.dao.janusgraph.JanusGraphOperationStatus;
  46 import org.openecomp.sdc.be.model.DataTypeDefinition;
  47 import org.openecomp.sdc.be.model.operations.impl.DataTypeOperation;
  48 import org.openecomp.sdc.be.model.operations.impl.PropertyOperation;
  49 import org.openecomp.sdc.be.resources.data.DataTypeData;
  50 import org.openecomp.sdc.common.log.enums.EcompLoggerErrorCode;
  51 import org.openecomp.sdc.common.log.wrappers.Logger;
  52 import org.springframework.context.ApplicationEvent;
  53 import org.springframework.context.ApplicationEventPublisher;
  54 import org.springframework.stereotype.Component;
  55
  56 @Component("application-datatype-cache")
  57 public class ApplicationDataTypeCache implements ApplicationCache<DataTypeDefinition>, Runnable {
  58
  59     private static final String APPLICATION_DATA_TYPES_CACHE = "ApplicationDataTypesCache";
  60     private static final Logger log = Logger.getLogger(ApplicationDataTypeCache.class);
  61
  62     private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  63     private final PropertyOperation propertyOperation;
  64     private final ApplicationEventPublisher applicationEventPublisher;
  65     @Getter(AccessLevel.PACKAGE)
  66     private final ScheduledExecutorService scheduledPollingService;
  67     @Getter(AccessLevel.PACKAGE)
  68     private ScheduledFuture<?> scheduledFuture = null;
  69     private Map<String, Map<String, DataTypeDefinition>> dataTypesByModelCacheMap = new HashMap<>();
  70     private final DataTypeOperation dataTypeOperation;
  71     private int firstRunDelayInSec = 30;
  72     private int pollingIntervalInSec = 60;
  73
  74     public ApplicationDataTypeCache(final PropertyOperation propertyOperation, final ApplicationEventPublisher applicationEventPublisher,
  75                                     final DataTypeOperation dataTypeOperation) {
  76         this.propertyOperation = propertyOperation;
  77         this.applicationEventPublisher = applicationEventPublisher;
  78         this.dataTypeOperation = dataTypeOperation;
  79         scheduledPollingService = Executors
  80             .newScheduledThreadPool(1, new BasicThreadFactory.Builder().namingPattern("ApplicationDataTypeCacheThread-%d").build());
  81     }
  82
  83     @PostConstruct
  84     void init() {
  85         final Optional<ApplicationL1CacheInfo> dataTypeCacheConfigOptional = getDataTypeCacheConfig();
  86         if (dataTypeCacheConfigOptional.isEmpty()) {
  87             BeEcompErrorManager.getInstance()
  88                 .logInternalFlowError(APPLICATION_DATA_TYPES_CACHE, "Data types cache is not configured and will be disabled", ErrorSeverity.INFO);
  89             return;
  90         }
  91         final ApplicationL1CacheInfo dataTypesCacheInfo = dataTypeCacheConfigOptional.get();
  92         if (!Boolean.TRUE.equals(dataTypesCacheInfo.getEnabled())) {
  93             BeEcompErrorManager.getInstance().logInternalFlowError(APPLICATION_DATA_TYPES_CACHE, "Data types cache is disabled", ErrorSeverity.INFO);
  94             return;
  95         }
  96         loadConfigurationValues(dataTypesCacheInfo);
  97         if (scheduledPollingService != null) {
  98             log.debug("Starting ApplicationDataTypeCache polling task. Initial delay {}s and polling interval {}s",
  99                 firstRunDelayInSec, pollingIntervalInSec);
 100             scheduledFuture = scheduledPollingService
 101                 .scheduleAtFixedRate(this, firstRunDelayInSec, pollingIntervalInSec, TimeUnit.SECONDS);
 102         }
 103     }
 104
 105     private void loadConfigurationValues(final ApplicationL1CacheInfo dataTypesCacheInfo) {
 106         final Integer firstRunDelay = dataTypesCacheInfo.getFirstRunDelay();
 107         if (firstRunDelay != null) {
 108             firstRunDelayInSec = firstRunDelay;
 109         }
 110         log.trace("ApplicationDataTypesCache initial delay configured to {} seconds.", firstRunDelayInSec);
 111
 112         final Integer intervalInSec = dataTypesCacheInfo.getPollIntervalInSec();
 113         if (intervalInSec != null) {
 114             pollingIntervalInSec = intervalInSec;
 115         }
 116         log.trace("ApplicationDataTypesCache polling interval configured to {} seconds.", pollingIntervalInSec);
 117     }
 118
 119     private Optional<ApplicationL1CacheInfo> getDataTypeCacheConfig() {
 120         final var applicationL1CacheConfig = ConfigurationManager.getConfigurationManager().getConfiguration().getApplicationL1Cache();
 121         if (applicationL1CacheConfig == null || applicationL1CacheConfig.getDatatypes() == null) {
 122             return Optional.empty();
 123         }
 124         return Optional.ofNullable(applicationL1CacheConfig.getDatatypes());
 125     }
 126
 127     @PreDestroy
 128     void destroy() {
 129         if (scheduledFuture != null) {
 130             boolean result = scheduledFuture.cancel(true);
 131             log.debug("Stop polling task. result = {}", result);
 132             scheduledFuture = null;
 133         }
 134         shutdownExecutor();
 135     }
 136
 137     private void shutdownExecutor() {
 138         if (scheduledPollingService == null) {
 139             return;
 140         }
 141         scheduledPollingService.shutdown(); // Disable new tasks from being
 142
 143         // submitted
 144         try {
 145             // Wait a while for existing tasks to terminate
 146             if (!scheduledPollingService.awaitTermination(60, TimeUnit.SECONDS)) {
 147                 scheduledPollingService.shutdownNow(); // Cancel currently
 148
 149                 // executing tasks
 150
 151                 // Wait a while for tasks to respond to being cancelled
 152                 if (!scheduledPollingService.awaitTermination(60, TimeUnit.SECONDS)) {
 153                     log.debug("Pool did not terminate");
 154                 }
 155             }
 156         } catch (InterruptedException ie) {
 157             // (Re-)Cancel if current thread also interrupted
 158             scheduledPollingService.shutdownNow();
 159             // Preserve interrupt status
 160             Thread.currentThread().interrupt();
 161         }
 162     }
 163
 164     private Either<Map<String, Map<String, DataTypeDefinition>>, JanusGraphOperationStatus> getAllDataTypesFromGraph() {
 165         return propertyOperation.getAllDataTypes();
 166     }
 167
 168     public Either<Map<String, DataTypeDefinition>, JanusGraphOperationStatus> getAll(final String model) {
 169         try {
 170             readWriteLock.readLock().lock();
 171             if (MapUtils.isEmpty(dataTypesByModelCacheMap) || !dataTypesByModelCacheMap.containsKey(model)) {
 172                 final var dataTypesFound = getAllDataTypesFromGraph();
 173                 if (dataTypesFound.isRight()) {
 174                     return Either.right(dataTypesFound.right().value());
 175                 }
 176                 dataTypesByModelCacheMap = dataTypesFound.left().value();
 177             }
 178             return Either.left(getDataTypeDefinitionMapByModel(model));
 179         } finally {
 180             readWriteLock.readLock().unlock();
 181         }
 182     }
 183
 184     @Override
 185     public Either<DataTypeDefinition, JanusGraphOperationStatus> get(final String model, final String uniqueId) {
 186         try {
 187             readWriteLock.readLock().lock();
 188             if (MapUtils.isEmpty(dataTypesByModelCacheMap)) {
 189                 return propertyOperation.getDataTypeByUid(uniqueId);
 190             }
 191             final Optional<DataTypeDefinition> dataTypeDefinition = getDataTypeDefinitionMapByModel(model).values().stream()
 192                 .filter(p -> p.getUniqueId().equals(uniqueId)).findFirst();
 193             if (dataTypeDefinition.isEmpty()) {
 194                 return propertyOperation.getDataTypeByUid(uniqueId);
 195             }
 196             return Either.left(new DataTypeDefinition(dataTypeDefinition.get()));
 197         } finally {
 198             readWriteLock.readLock().unlock();
 199         }
 200     }
 201
 202     public void reload(final String model, final String uniqueId) {
 203         final Either<DataTypeDefinition, JanusGraphOperationStatus> dataTypeDefEither = propertyOperation.getDataTypeByUid(uniqueId);
 204         if (dataTypeDefEither.isLeft()) {
 205             DataTypeDefinition dataTypeDef = dataTypeDefEither.left().value();
 206             if (getDataTypeDefinitionMapByModel(model).containsKey(dataTypeDef.getName())) {
 207                 try {
 208                     readWriteLock.readLock().lock();
 209                     getDataTypeDefinitionMapByModel(model).put(dataTypeDef.getName(), dataTypeDef);
 210                 } finally {
 211                     readWriteLock.readLock().unlock();
 212                 }
 213             }
 214         }
 215     }
 216
 217     private Map<String, DataTypeDefinition> getDataTypeDefinitionMapByModel(final String model) {
 218         return dataTypesByModelCacheMap.containsKey(model) ? dataTypesByModelCacheMap.get(model) : new HashMap<>();
 219     }
 220
 221     @Override
 222     public void run() {
 223         try {
 224             refreshDataTypesCacheIfStale();
 225         } catch (final Exception e) {
 226             var errorMsg = "Failed to run refresh data types cache job";
 227             log.error(EcompLoggerErrorCode.UNKNOWN_ERROR, ApplicationDataTypeCache.class.getName(), errorMsg, e);
 228             BeEcompErrorManager.getInstance().logInternalUnexpectedError(APPLICATION_DATA_TYPES_CACHE, errorMsg, ErrorSeverity.INFO);
 229         } finally {
 230             try {
 231                 propertyOperation.getJanusGraphGenericDao().commit();
 232             } catch (final Exception e) {
 233                 log.error(EcompLoggerErrorCode.UNKNOWN_ERROR, ApplicationDataTypeCache.class.getName(),
 234                     "Failed to commit ApplicationDataTypeCache", e);
 235             }
 236         }
 237     }
 238
 239     private boolean hasDataTypesChanged() {
 240         final List<DataTypeData> dataTypeListFromDatabase = findAllDataTypesLazy();
 241         final int dataTypesCacheCopyMap = dataTypesCacheMapSize();
 242         if (dataTypeListFromDatabase.size() != dataTypesCacheCopyMap) {
 243             log.debug("Total of cached data types '{}' differs from the actual '{}'", dataTypeListFromDatabase.size(),  dataTypesCacheCopyMap);
 244             return true;
 245         }
 246
 247         if (CollectionUtils.isEmpty(dataTypeListFromDatabase)) {
 248             log.debug("Both data type cache and database are empty");
 249             return false;
 250         }
 251
 252         return hasDataTypesChanged(dataTypeListFromDatabase, copyDataTypeCache());
 253     }
 254
 255     private int dataTypesCacheMapSize() {
 256         var count = 0;
 257         for (var i = 0; i < copyDataTypeCache().size(); i++) {
 258             count += new ArrayList<>(copyDataTypeCache().values()).get(i).size();
 259
 260         }
 261         return count;
 262     }
 263
 264     private boolean hasDataTypesChanged(final List<DataTypeData> dataTypeListFromDatabase, final Map<String, Map<String, DataTypeDefinition>> dataTypesCacheCopyMap) {
 265         return dataTypeListFromDatabase.stream().map(DataTypeData::getDataTypeDataDefinition).anyMatch(actualDataTypeDefinition -> {
 266             final String dataTypeName = actualDataTypeDefinition.getName();
 267             final String model = actualDataTypeDefinition.getModel();
 268             final DataTypeDefinition cachedDataTypeDefinition = dataTypesCacheCopyMap.get(model).get(dataTypeName);
 269             if (cachedDataTypeDefinition == null) {
 270                 log.debug("Datatype '{}' is not present in the cache. ", dataTypeName);
 271                 return true;
 272             }
 273
 274             final long cachedCreationTime = cachedDataTypeDefinition.getCreationTime() == null ? 0 : cachedDataTypeDefinition.getCreationTime();
 275             final long actualCreationTime = actualDataTypeDefinition.getCreationTime() == null ? 0 : actualDataTypeDefinition.getCreationTime();
 276             if (cachedCreationTime != actualCreationTime) {
 277                 log.debug("Datatype '{}' with model '{}' was updated. Cache/database creation time '{}'/'{}'.",
 278                     dataTypeName, model, cachedCreationTime, actualCreationTime);
 279                 return true;
 280             }
 281             final long cachedModificationTime =
 282                 cachedDataTypeDefinition.getModificationTime() == null ? 0 : cachedDataTypeDefinition.getModificationTime();
 283             final long actualModificationTime =
 284                 actualDataTypeDefinition.getModificationTime() == null ? 0 : actualDataTypeDefinition.getModificationTime();
 285             if (cachedModificationTime != actualModificationTime) {
 286                 log.debug("Datatype '{}' was updated. Cache/database modification time '{}'/'{}'.",
 287                     dataTypeName, cachedModificationTime, actualModificationTime);
 288                 return true;
 289             }
 290
 291             return false;
 292         });
 293     }
 294
 295     private Map<String, Map<String, DataTypeDefinition>> copyDataTypeCache() {
 296         try {
 297             readWriteLock.readLock().lock();
 298             return new HashMap<>(this.dataTypesByModelCacheMap);
 299         } finally {
 300             readWriteLock.readLock().unlock();
 301         }
 302     }
 303
 304     public void refreshDataTypesCacheIfStale() {
 305         final long startTime = System.currentTimeMillis();
 306         log.trace("Starting refresh data types cache");
 307         if (hasDataTypesChanged()) {
 308             log.info("Detected changes in the data types, updating the data type cache.");
 309             refreshDataTypesCache();
 310         }
 311         log.trace("Finished refresh data types cache. Finished in {}ms", (System.currentTimeMillis() - startTime));
 312     }
 313
 314     private void refreshDataTypesCache() {
 315         final Map<String, Map<String, DataTypeDefinition>> dataTypesDefinitionMap = findAllDataTypesEager();
 316         if (dataTypesDefinitionMap.isEmpty()) {
 317             return;
 318         }
 319         try {
 320             readWriteLock.writeLock().lock();
 321             dataTypesByModelCacheMap = dataTypesDefinitionMap;
 322             onDataChangeEventEmit();
 323             BeEcompErrorManager.getInstance()
 324                 .logInternalFlowError("ReplaceDataTypesCache", "Succeed to replace the data types cache", ErrorSeverity.INFO);
 325         } finally {
 326             readWriteLock.writeLock().unlock();
 327         }
 328     }
 329
 330     private Map<String, Map<String, DataTypeDefinition>> findAllDataTypesEager() {
 331         log.trace("Fetching data types from database, eager mode");
 332         final long startTime = System.currentTimeMillis();
 333         final Either<Map<String, Map<String, DataTypeDefinition>>, JanusGraphOperationStatus> allDataTypes = propertyOperation.getAllDataTypes();
 334         log.trace("Finish fetching data types from database. Took {}ms", (System.currentTimeMillis() - startTime));
 335         if (allDataTypes.isRight()) {
 336             final JanusGraphOperationStatus status = allDataTypes.right().value();
 337             var errorMsg= String.format("Failed to fetch data types from database. Status is %s", status);
 338             log.error(EcompLoggerErrorCode.UNKNOWN_ERROR, ApplicationDataTypeCache.class.getName(), errorMsg);
 339             BeEcompErrorManager.getInstance().logInternalConnectionError(APPLICATION_DATA_TYPES_CACHE, errorMsg, ErrorSeverity.ERROR);
 340             return Collections.emptyMap();
 341         }
 342         return allDataTypes.left().value();
 343     }
 344
 345     private List<DataTypeData> findAllDataTypesLazy() {
 346         log.trace("Fetching data types from database, lazy mode");
 347         final long startTime = System.currentTimeMillis();
 348         final List<DataTypeData> allDataTypes = dataTypeOperation.getAllDataTypeNodes();
 349         log.trace("Finish fetching data types from database. Took {}ms", (System.currentTimeMillis() - startTime));
 350         return allDataTypes;
 351     }
 352
 353     private void onDataChangeEventEmit() {
 354         log.trace("Data type cache has changed, sending DataTypesCacheChangedEvent.");
 355         applicationEventPublisher.publishEvent(new DataTypesCacheChangedEvent(this, copyDataTypeCache()));
 356     }
 357
 358     /**
 359      * Custom event to notify all interested in cached data changes
 360      */
 361     public static class DataTypesCacheChangedEvent extends ApplicationEvent {
 362
 363         @Getter
 364         private final Map<String,  Map<String, DataTypeDefinition>> newData;
 365
 366         public DataTypesCacheChangedEvent(final Object source, final Map<String,  Map<String, DataTypeDefinition>> newData) {
 367             super(source);
 368             this.newData = newData;
 369         }
 370     }
 371
 372 }