diff options
Diffstat (limited to 'cps-ncmp-service')
4 files changed, 27 insertions, 18 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java index d457f2601b..d5b459b025 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java @@ -56,7 +56,7 @@ public class DmiRestClient { } catch (final HttpStatusCodeException httpStatusCodeException) { final String exceptionMessage = "Unable to " + operation.toString() + " resource data."; throw new HttpClientRequestException(exceptionMessage, httpStatusCodeException.getResponseBodyAsString(), - httpStatusCodeException.getRawStatusCode()); + httpStatusCodeException.getRawStatusCode()); } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncTasks.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncTasks.java index ada3dc6744..f914547a50 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncTasks.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncTasks.java @@ -71,6 +71,7 @@ public class ModuleSyncTasks { moduleSyncService.syncAndCreateSchemaSetAndAnchor(yangModelCmHandle); cmHandelStatePerCmHandle.put(yangModelCmHandle, CmHandleState.READY); } catch (final Exception e) { + log.warn("Processing module sync batch failed."); syncUtils.updateLockReasonDetailsAndAttempts(compositeState, LockReasonCategory.LOCKED_MODULE_SYNC_FAILED, e.getMessage()); setCmHandleStateLocked(yangModelCmHandle, compositeState.getLockReason()); @@ -81,6 +82,7 @@ public class ModuleSyncTasks { lcmEventsCmHandleStateHandler.updateCmHandleStateBatch(cmHandelStatePerCmHandle); } finally { batchCounter.getAndDecrement(); + log.info("Processing module sync batch finished. {} batch(es) active.", batchCounter.get()); } return COMPLETED_FUTURE; } 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 73954c36b3..64d111f993 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 @@ -62,16 +62,22 @@ public class ModuleSyncWatchdog { */ @Scheduled(fixedDelayString = "${timers.advised-modules-sync.sleep-time-ms:5000}") public void moduleSyncAdvisedCmHandles() { + log.info("Processing module sync watchdog waking up."); populateWorkQueueIfNeeded(); final int asyncTaskParallelismLevel = asyncTaskExecutor.getAsyncTaskParallelismLevel(); - while (!moduleSyncWorkQueue.isEmpty() && batchCounter.get() <= asyncTaskParallelismLevel) { - batchCounter.getAndIncrement(); - final Collection<DataNode> nextBatch = prepareNextBatch(); - asyncTaskExecutor.executeTask(() -> - moduleSyncTasks.performModuleSync(nextBatch, batchCounter), - ASYNC_TASK_TIMEOUT_IN_MILLISECONDS - ); - preventBusyWait(); + while (!moduleSyncWorkQueue.isEmpty()) { + if (batchCounter.get() <= asyncTaskParallelismLevel) { + final Collection<DataNode> nextBatch = prepareNextBatch(); + log.debug("Processing module sync batch of {}. {} batch(es) active.", + nextBatch.size(), batchCounter.get()); + asyncTaskExecutor.executeTask(() -> + moduleSyncTasks.performModuleSync(nextBatch, batchCounter), + ASYNC_TASK_TIMEOUT_IN_MILLISECONDS + ); + batchCounter.getAndIncrement(); + } else { + preventBusyWait(); + } } } @@ -80,6 +86,7 @@ public class ModuleSyncWatchdog { */ @Scheduled(fixedDelayString = "${timers.locked-modules-sync.sleep-time-ms:300000}") public void resetPreviouslyFailedCmHandles() { + log.info("Processing module sync retry-watchdog waking up."); final List<YangModelCmHandle> failedCmHandles = syncUtils.getModuleSyncFailedCmHandles(); moduleSyncTasks.resetFailedCmHandles(failedCmHandles); } @@ -95,6 +102,7 @@ public class ModuleSyncWatchdog { private void populateWorkQueueIfNeeded() { if (moduleSyncWorkQueue.isEmpty()) { final List<DataNode> advisedCmHandles = syncUtils.getAdvisedCmHandles(); + log.info("Processing module sync fetched {} advised cm handles from DB", advisedCmHandles.size()); for (final DataNode advisedCmHandle : advisedCmHandles) { if (!moduleSyncWorkQueue.offer(advisedCmHandle)) { log.warn("Unable to add cm handle {} to the work queue", advisedCmHandle.getLeaves().get("id")); diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdogSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdogSpec.groovy index e5240c0e66..dd989bf676 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdogSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdogSpec.groovy @@ -52,20 +52,19 @@ class ModuleSyncWatchdogSpec extends Specification { def 'Module sync advised cm handles with #scenario.'() { given: 'sync utilities returns #numberOfAdvisedCmHandles advised cm handles' mockSyncUtils.getAdvisedCmHandles() >> createDataNodes(numberOfAdvisedCmHandles) - and: 'the executor has #parallelismLevel available threads' - spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >> parallelismLevel + 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 | parallelismLevel | numberOfAdvisedCmHandles || expectedNumberOfTaskExecutions - 'less then 1 batch' | 9 | 1 || 1 - 'exactly 1 batch' | 9 | ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 1 - '2 batches' | 9 | 2 * ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 2 - 'queue capacity' | 9 | testQueueCapacity || 3 - 'over queue capacity' | 9 | testQueueCapacity + 2 * ModuleSyncWatchdog.MODULE_SYNC_BATCH_SIZE || 3 - 'not enough threads' | 2 | testQueueCapacity || 2 + 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 'Reset failed cm handles.'() { |