diff options
Diffstat (limited to 'cps-ncmp-service')
5 files changed, 53 insertions, 17 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfig.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfig.java index c89388b291..5154be7990 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfig.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfig.java @@ -28,7 +28,6 @@ import com.hazelcast.core.Hazelcast; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.map.IMap; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.TimeUnit; import org.onap.cps.spi.model.DataNode; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -39,11 +38,12 @@ import org.springframework.context.annotation.Configuration; @Configuration public class SynchronizationCacheConfig { + public static final int MODULE_SYNC_STARTED_TTL_SECS = 60; + public static final int DATA_SYNC_SEMAPHORE_TTL_SECS = 1800; + private static final QueueConfig commonQueueConfig = createQueueConfig(); - private static final MapConfig moduleSyncStartedConfig = - createMapConfig("moduleSyncStartedConfig", TimeUnit.MINUTES.toSeconds(1)); - private static final MapConfig dataSyncSemaphoresConfig = - createMapConfig("dataSyncSemaphoresConfig", TimeUnit.MINUTES.toSeconds(30)); + private static final MapConfig moduleSyncStartedConfig = createMapConfig("moduleSyncStartedConfig"); + private static final MapConfig dataSyncSemaphoresConfig = createMapConfig("dataSyncSemaphoresConfig"); /** * Module Sync Distributed Queue Instance. @@ -102,11 +102,10 @@ public class SynchronizationCacheConfig { return commonQueueConfig; } - private static MapConfig createMapConfig(final String configName, final long timeToLiveSeconds) { + private static MapConfig createMapConfig(final String configName) { final MapConfig mapConfig = new MapConfig(configName); mapConfig.setBackupCount(3); mapConfig.setAsyncBackupCount(3); - mapConfig.setTimeToLiveSeconds((int) timeToLiveSeconds); return mapConfig; } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdog.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdog.java index 9336c3b218..9fa75a0d07 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdog.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdog.java @@ -20,12 +20,14 @@ package org.onap.cps.ncmp.api.inventory.sync; +import com.hazelcast.map.IMap; import java.time.OffsetDateTime; -import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.onap.cps.api.CpsDataService; +import org.onap.cps.ncmp.api.impl.config.embeddedcache.SynchronizationCacheConfig; import org.onap.cps.ncmp.api.inventory.CompositeState; import org.onap.cps.ncmp.api.inventory.DataStoreSyncState; import org.onap.cps.ncmp.api.inventory.InventoryPersistence; @@ -46,7 +48,7 @@ public class DataSyncWatchdog { private final SyncUtils syncUtils; - private final Map<String, Boolean> dataSyncSemaphores; + private final IMap<String, Boolean> dataSyncSemaphores; /** * Execute Cm Handle poll which queries the cm handle state in 'READY' and Operational Datastore Sync State in @@ -92,6 +94,7 @@ public class DataSyncWatchdog { } private boolean hasPushedIntoSemaphoreMap(final String cmHandleId) { - return dataSyncSemaphores.putIfAbsent(cmHandleId, DATA_SYNC_IN_PROGRESS) == null; + return dataSyncSemaphores.putIfAbsent(cmHandleId, DATA_SYNC_IN_PROGRESS, + SynchronizationCacheConfig.DATA_SYNC_SEMAPHORE_TTL_SECS, TimeUnit.SECONDS) == null; } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java index b96889fc58..f629b71d26 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java @@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicInteger; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.api.impl.config.embeddedcache.SynchronizationCacheConfig; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle; import org.onap.cps.ncmp.api.inventory.sync.executor.AsyncTaskExecutor; import org.onap.cps.spi.model.DataNode; @@ -117,8 +118,9 @@ public class ModuleSyncWatchdog { log.debug("nextBatchCandidates size : {}", nextBatchCandidates.size()); for (final DataNode batchCandidate : nextBatchCandidates) { final String cmHandleId = String.valueOf(batchCandidate.getLeaves().get("id")); - final boolean alreadyAddedToInProgressMap = VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP - .equals(moduleSyncStartedOnCmHandles.putIfAbsent(cmHandleId, VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP)); + final boolean alreadyAddedToInProgressMap = VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP.equals( + moduleSyncStartedOnCmHandles.putIfAbsent(cmHandleId, VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP, + SynchronizationCacheConfig.MODULE_SYNC_STARTED_TTL_SECS, TimeUnit.SECONDS)); if (alreadyAddedToInProgressMap) { log.debug("module sync for {} already in progress by other instance", cmHandleId); } else { diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfigSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfigSpec.groovy index 4cfc02b9e7..c16d6b69b6 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfigSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/config/embeddedcache/SynchronizationCacheConfigSpec.groovy @@ -28,6 +28,7 @@ import org.springframework.boot.test.context.SpringBootTest import org.springframework.test.context.ContextConfiguration import spock.lang.Specification import java.util.concurrent.BlockingQueue +import java.util.concurrent.TimeUnit @SpringBootTest @ContextConfiguration(classes = [SynchronizationCacheConfig]) @@ -40,7 +41,7 @@ class SynchronizationCacheConfigSpec extends Specification { private IMap<String, Object> moduleSyncStartedOnCmHandles @Autowired - private Map<String, Boolean> dataSyncSemaphores + private IMap<String, Boolean> dataSyncSemaphores def 'Embedded (hazelcast) Caches for Module and Data Sync.'() { expect: 'system is able to create an instance of the Module Sync Work Queue' @@ -54,4 +55,36 @@ class SynchronizationCacheConfigSpec extends Specification { and: 'they have the correct names (in any order)' assert Hazelcast.allHazelcastInstances.name.containsAll('moduleSyncWorkQueue', 'moduleSyncStartedOnCmHandles', 'dataSyncSemaphores' ) } + + def 'Verify configs for Distributed objects'(){ + given: 'the Module Sync Work Queue config' + def queueConfig = Hazelcast.getHazelcastInstanceByName('moduleSyncWorkQueue').config.queueConfigs.get('defaultQueueConfig') + and: 'the Module Sync Started Cm Handle Map config' + def moduleSyncStartedOnCmHandlesConfig = Hazelcast.getHazelcastInstanceByName('moduleSyncStartedOnCmHandles').config.mapConfigs.get('moduleSyncStartedConfig') + and: 'the Data Sync Semaphores Map config' + def dataSyncSemaphoresConfig = Hazelcast.getHazelcastInstanceByName('dataSyncSemaphores').config.mapConfigs.get('dataSyncSemaphoresConfig') + expect: 'system created instance with correct config of Module Sync Work Queue' + assert queueConfig.backupCount == 3 + assert queueConfig.asyncBackupCount == 3 + and: 'Module Sync Started Cm Handle Map has the correct settings' + assert moduleSyncStartedOnCmHandlesConfig.backupCount == 3 + assert moduleSyncStartedOnCmHandlesConfig.asyncBackupCount == 3 + and: 'Data Sync Semaphore Map has the correct settings' + assert dataSyncSemaphoresConfig.backupCount == 3 + assert dataSyncSemaphoresConfig.asyncBackupCount == 3 + } + + def 'Time to Live Verify for Module Sync and Data Sync Semaphore'() { + when: 'the keys are inserted with a TTL' + moduleSyncStartedOnCmHandles.put('testKeyModuleSync', 'toBeExpired' as Object, 1000, TimeUnit.MILLISECONDS) + dataSyncSemaphores.put('testKeyDataSync', Boolean.TRUE, 1000, TimeUnit.MILLISECONDS) + then: 'the entries are present in the map' + assert moduleSyncStartedOnCmHandles.get('testKeyModuleSync') != null + assert dataSyncSemaphores.get('testKeyDataSync') != null + and: 'we wait for the key expiration' + sleep(1500) + and: 'the keys should be expired as TTL elapsed' + assert moduleSyncStartedOnCmHandles.get('testKeyModuleSync') == null + assert dataSyncSemaphores.get('testKeyDataSync') == null + } } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdogSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdogSpec.groovy index 605381970d..707f3ea3ea 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdogSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/DataSyncWatchdogSpec.groovy @@ -20,6 +20,7 @@ package org.onap.cps.ncmp.api.inventory.sync +import com.hazelcast.map.IMap import org.onap.cps.api.CpsDataService import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle import org.onap.cps.ncmp.api.inventory.CmHandleState @@ -27,8 +28,6 @@ import org.onap.cps.ncmp.api.inventory.CompositeState import org.onap.cps.ncmp.api.inventory.InventoryPersistence import org.onap.cps.ncmp.api.inventory.DataStoreSyncState import spock.lang.Specification -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.ConcurrentMap class DataSyncWatchdogSpec extends Specification { @@ -38,11 +37,11 @@ class DataSyncWatchdogSpec extends Specification { def mockSyncUtils = Mock(SyncUtils) - def stubbedMap = Stub(ConcurrentMap) + def mockDataSyncSemaphoreMap = Mock(IMap<String,Boolean>) def jsonString = '{"stores:bookstore":{"categories":[{"code":"01"}]}}' - def objectUnderTest = new DataSyncWatchdog(mockInventoryPersistence, mockCpsDataService, mockSyncUtils, stubbedMap as ConcurrentHashMap) + def objectUnderTest = new DataSyncWatchdog(mockInventoryPersistence, mockCpsDataService, mockSyncUtils, mockDataSyncSemaphoreMap) def compositeState = getCompositeState() |