/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation
+ * Copyright (C) 2022-2023 Nordix Foundation
* Modifications Copyright (C) 2022 Bell Canada
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
package org.onap.cps.ncmp.api.inventory.sync
-
-import org.onap.cps.ncmp.api.impl.event.lcm.LcmEventsCmHandleStateHandler
+import com.hazelcast.map.IMap
+import org.onap.cps.ncmp.api.impl.inventory.sync.ModuleSyncTasks
+import org.onap.cps.ncmp.api.impl.inventory.sync.ModuleSyncWatchdog
+import org.onap.cps.ncmp.api.impl.inventory.sync.SyncUtils
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
-import org.onap.cps.ncmp.api.inventory.CmHandleState
-import org.onap.cps.ncmp.api.inventory.CompositeState
-import org.onap.cps.ncmp.api.inventory.InventoryPersistence
-import org.onap.cps.ncmp.api.inventory.LockReasonCategory
-import org.onap.cps.ncmp.api.inventory.CompositeStateBuilder
+import org.onap.cps.ncmp.api.impl.inventory.sync.executor.AsyncTaskExecutor
+import java.util.concurrent.ArrayBlockingQueue
+import org.onap.cps.spi.model.DataNode
import spock.lang.Specification
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentMap
-
class ModuleSyncWatchdogSpec extends Specification {
- def mockInventoryPersistence = Mock(InventoryPersistence)
-
def mockSyncUtils = Mock(SyncUtils)
- def mockModuleSyncService = Mock(ModuleSyncService)
-
- def stubbedMap = Stub(ConcurrentMap)
-
- def mockLcmEventsCmHandleStateHandler = Mock(LcmEventsCmHandleStateHandler)
-
- def cmHandleState = CmHandleState.ADVISED
-
- def objectUnderTest = new ModuleSyncWatchdog(mockInventoryPersistence, mockSyncUtils, mockModuleSyncService, stubbedMap as ConcurrentHashMap, mockLcmEventsCmHandleStateHandler)
-
- def 'Schedule a Cm-Handle Sync for ADVISED Cm-Handles'() {
- given: 'cm handles in an advised state and a data sync state'
- def compositeState1 = new CompositeState(cmHandleState: cmHandleState)
- def compositeState2 = new CompositeState(cmHandleState: cmHandleState)
- def yangModelCmHandle1 = new YangModelCmHandle(id: 'some-cm-handle', compositeState: compositeState1)
- def yangModelCmHandle2 = new YangModelCmHandle(id: 'some-cm-handle-2', compositeState: compositeState2)
- and: 'sync utilities return a cm handle twice'
- mockSyncUtils.getAdvisedCmHandles() >> [yangModelCmHandle1, yangModelCmHandle2]
- when: 'module sync poll is executed'
- objectUnderTest.executeAdvisedCmHandlePoll()
- then: 'the inventory persistence cm handle returns a composite state for the first cm handle'
- 1 * mockInventoryPersistence.getCmHandleState('some-cm-handle') >> compositeState1
- and: 'module sync service deletes schema set of cm handle if it exists'
- 1 * mockModuleSyncService.deleteSchemaSetIfExists(yangModelCmHandle1)
- and: 'module sync service syncs the first cm handle and creates a schema set'
- 1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(yangModelCmHandle1)
- then: 'the state handler is called for the first cm handle'
- 1 * mockLcmEventsCmHandleStateHandler.updateCmHandleState(yangModelCmHandle1, CmHandleState.READY)
- and: 'the inventory persistence cm handle returns a composite state for the second cm handle'
- mockInventoryPersistence.getCmHandleState('some-cm-handle-2') >> compositeState2
- and: 'module sync service syncs the second cm handle and creates a schema set'
- 1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(yangModelCmHandle2)
- then: 'the state handler is called for the second cm handle'
- 1 * mockLcmEventsCmHandleStateHandler.updateCmHandleState(yangModelCmHandle2, CmHandleState.READY)
+ def static testQueueCapacity = 50 + 2 * ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE
+
+ def moduleSyncWorkQueue = new ArrayBlockingQueue(testQueueCapacity)
+
+ def mockModuleSyncStartedOnCmHandles = Mock(IMap<String, Object>)
+
+ def mockModuleSyncTasks = Mock(ModuleSyncTasks)
+
+ def spiedAsyncTaskExecutor = Spy(AsyncTaskExecutor)
+
+ def moduleSetTagCache = Mock(IMap<String, Set<String>>)
+
+ def objectUnderTest = new ModuleSyncWatchdog(mockSyncUtils, moduleSyncWorkQueue , mockModuleSyncStartedOnCmHandles, mockModuleSyncTasks, spiedAsyncTaskExecutor, moduleSetTagCache)
+
+ void setup() {
+ spiedAsyncTaskExecutor.setupThreadPool()
+ }
+
+ def 'Module sync advised cm handles with #scenario.'() {
+ given: 'sync utilities returns #numberOfAdvisedCmHandles advised cm handles'
+ mockSyncUtils.getAdvisedCmHandles() >> createDataNodes(numberOfAdvisedCmHandles)
+ and: 'the executor has enough available threads'
+ spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >> 3
+ when: ' module sync is started'
+ objectUnderTest.moduleSyncAdvisedCmHandles()
+ then: 'it performs #expectedNumberOfTaskExecutions tasks'
+ expectedNumberOfTaskExecutions * spiedAsyncTaskExecutor.executeTask(*_)
+ where: 'the following parameter are used'
+ scenario | numberOfAdvisedCmHandles || expectedNumberOfTaskExecutions
+ 'less then 1 batch' | 1 || 1
+ 'exactly 1 batch' | ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 1
+ '2 batches' | 2 * ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 2
+ 'queue capacity' | testQueueCapacity || 3
+ 'over queue capacity' | testQueueCapacity + 2 * ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 3
+ }
+
+ def 'Module sync advised cm handles starts with no available threads.'() {
+ given: 'sync utilities returns a advise cm handles'
+ mockSyncUtils.getAdvisedCmHandles() >> createDataNodes(1)
+ and: 'the executor first has no threads but has one thread on the second attempt'
+ spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >>> [ 0, 1 ]
+ when: ' module sync is started'
+ objectUnderTest.moduleSyncAdvisedCmHandles()
+ then: 'it performs one task'
+ 1 * spiedAsyncTaskExecutor.executeTask(*_)
+ }
+
+ def 'Module sync advised cm handles already handled.'() {
+ given: 'sync utilities returns a advise cm handles'
+ mockSyncUtils.getAdvisedCmHandles() >> createDataNodes(1)
+ and: 'the executor has a thread available'
+ spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >> 1
+ and: 'the semaphore cache indicates the cm handle is already being processed'
+ mockModuleSyncStartedOnCmHandles.putIfAbsent(*_) >> 'Started'
+ when: ' module sync is started'
+ objectUnderTest.moduleSyncAdvisedCmHandles()
+ then: 'it does NOT execute a task to process the (empty) batch'
+ 0 * spiedAsyncTaskExecutor.executeTask(*_)
+ }
+
+ def 'Module sync with previous cm handle(s) left in work queue.'() {
+ given: 'there is still a cm handle in the queue'
+ moduleSyncWorkQueue.offer(new DataNode())
+ and: 'sync utilities returns many advise cm handles'
+ mockSyncUtils.getAdvisedCmHandles() >> createDataNodes(500)
+ and: 'the executor has plenty threads available'
+ spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >> 10
+ when: ' module sync is started'
+ objectUnderTest.moduleSyncAdvisedCmHandles()
+ then: 'it does executes only one task to process the remaining handle in the queue'
+ 1 * spiedAsyncTaskExecutor.executeTask(*_)
}
- def 'Schedule a Cm-Handle Sync for ADVISED Cm-Handle with failure'() {
- given: 'cm handles in an advised state'
- def compositeState = new CompositeState(cmHandleState: cmHandleState)
- def yangModelCmHandle = new YangModelCmHandle(id: 'some-cm-handle', compositeState: compositeState)
- and: 'sync utilities return a cm handle'
- mockSyncUtils.getAdvisedCmHandles() >> [yangModelCmHandle]
- when: 'module sync poll is executed'
- objectUnderTest.executeAdvisedCmHandlePoll()
- then: 'the inventory persistence cm handle returns a composite state for the cm handle'
- 1 * mockInventoryPersistence.getCmHandleState('some-cm-handle') >> compositeState
- and: 'module sync service attempts to sync the cm handle and throws an exception'
- 1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(*_) >> { throw new Exception('some exception') }
- and: 'update lock reason, details and attempts is invoked'
- 1 * mockSyncUtils.updateLockReasonDetailsAndAttempts(compositeState, LockReasonCategory.LOCKED_MODULE_SYNC_FAILED ,'some exception')
- and: 'the state handler is called to update the state to LOCKED'
- 1 * mockLcmEventsCmHandleStateHandler.updateCmHandleState(yangModelCmHandle, CmHandleState.LOCKED)
+ def 'Reset failed cm handles.'() {
+ given: 'sync utilities returns failed cm handles'
+ def failedCmHandles = [new YangModelCmHandle()]
+ mockSyncUtils.getCmHandlesThatFailedModelSyncOrUpgrade() >> failedCmHandles
+ when: 'reset failed cm handles is started'
+ objectUnderTest.resetPreviouslyFailedCmHandles()
+ then: 'it is delegated to the module sync task (service)'
+ 1 * mockModuleSyncTasks.resetFailedCmHandles(failedCmHandles)
}
- def 'Schedule a Cm-Handle Sync with condition #scenario '() {
- given: 'cm handles in an locked state'
- def compositeState = new CompositeStateBuilder().withCmHandleState(CmHandleState.LOCKED)
- .withLockReason(LockReasonCategory.LOCKED_MODULE_SYNC_FAILED, '').withLastUpdatedTimeNow().build()
- def yangModelCmHandle = new YangModelCmHandle(id: 'some-cm-handle', compositeState: compositeState)
- and: 'sync utilities return a cm handle twice'
- mockSyncUtils.getModuleSyncFailedCmHandles() >> [yangModelCmHandle, yangModelCmHandle]
- and: 'inventory persistence returns the composite state of the cm handle'
- mockInventoryPersistence.getCmHandleState(yangModelCmHandle.getId()) >> compositeState
- and: 'sync utils retry locked cm handle returns #isReadyForRetry'
- mockSyncUtils.isReadyForRetry(compositeState) >>> isReadyForRetry
- when: 'module sync poll is executed'
- objectUnderTest.executeLockedCmHandlePoll()
- then: 'the first cm handle is updated to state "ADVISED" from "READY"'
- expectedNumberOfInvocationsToSaveCmHandleState * mockLcmEventsCmHandleStateHandler.updateCmHandleState(yangModelCmHandle, CmHandleState.ADVISED)
- where:
- scenario | isReadyForRetry || expectedNumberOfInvocationsToSaveCmHandleState
- 'retry locked cm handle once' | [true, false] || 1
- 'retry locked cm handle twice' | [true, true] || 2
- 'do not retry locked cm handle' | [false, false] || 0
+ def createDataNodes(numberOfDataNodes) {
+ def dataNodes = []
+ (1..numberOfDataNodes).each {dataNodes.add(new DataNode())}
+ return dataNodes
}
}
Code Review / cps.git / blobdiff