diff options
Diffstat (limited to 'cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java')
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java | 119 |
1 files changed, 59 insertions, 60 deletions
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 be811a114..8074fe6fe 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 @@ -21,17 +21,16 @@ package org.onap.cps.ncmp.api.inventory.sync; +import java.util.Collection; +import java.util.HashSet; import java.util.List; -import java.util.concurrent.ConcurrentMap; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.onap.cps.ncmp.api.impl.event.lcm.LcmEventsCmHandleStateHandler; 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.springframework.beans.factory.annotation.Qualifier; +import org.onap.cps.spi.model.DataNode; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -40,75 +39,75 @@ import org.springframework.stereotype.Component; @Component public class ModuleSyncWatchdog { - private static final boolean MODEL_SYNC_IN_PROGRESS = false; - private static final boolean MODEL_SYNC_DONE = true; - - private final InventoryPersistence inventoryPersistence; - private final SyncUtils syncUtils; + private final BlockingQueue<DataNode> moduleSyncWorkQueue; + private final Map<String, Object> moduleSyncStartedOnCmHandles; + private final ModuleSyncTasks moduleSyncTasks; - private final ModuleSyncService moduleSyncService; - - @Qualifier("moduleSyncSemaphores") - private final ConcurrentMap<String, Boolean> moduleSyncSemaphores; - - private final LcmEventsCmHandleStateHandler lcmEventsCmHandleStateHandler; + private static final int MODULE_SYNC_BATCH_SIZE = 100; + private static final long PREVENT_CPU_BURN_WAIT_TIME_MILLIS = 10; + private static final String VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP = "Started"; /** * Execute Cm Handle poll which changes the cm handle state from 'ADVISED' to 'READY'. */ - @Scheduled(fixedDelayString = "${timers.advised-modules-sync.sleep-time-ms:30000}") - public void executeAdvisedCmHandlePoll() { - syncUtils.getAdvisedCmHandles().forEach(advisedCmHandle -> { - final String cmHandleId = advisedCmHandle.getId(); - if (hasPushedIntoSemaphoreMap(cmHandleId)) { - log.debug("executing module sync on {}", cmHandleId); - final CompositeState compositeState = inventoryPersistence.getCmHandleState(cmHandleId); - try { - moduleSyncService.deleteSchemaSetIfExists(advisedCmHandle); - moduleSyncService.syncAndCreateSchemaSetAndAnchor(advisedCmHandle); - lcmEventsCmHandleStateHandler.updateCmHandleState(advisedCmHandle, CmHandleState.READY); - updateModuleSyncSemaphoreMap(cmHandleId); - } catch (final Exception e) { - syncUtils.updateLockReasonDetailsAndAttempts(compositeState, - LockReasonCategory.LOCKED_MODULE_SYNC_FAILED, e.getMessage()); - setCmHandleStateLocked(advisedCmHandle, compositeState.getLockReason()); - } - log.debug("{} is now in {} state", cmHandleId, compositeState.getCmHandleState().name()); - } else { - log.debug("{} already processed by another instance", cmHandleId); - } - }); - log.debug("No Cm-Handles currently found in an ADVISED state"); + @Scheduled(fixedDelayString = "${timers.advised-modules-sync.sleep-time-ms:5000}") + public void moduleSyncAdvisedCmHandles() { + populateWorkQueueIfNeeded(); + while (!moduleSyncWorkQueue.isEmpty()) { + final Collection<DataNode> nextBatch = prepareNextBatch(); + moduleSyncTasks.performModuleSync(nextBatch); + preventBusyWait(); + } } /** - * Execute Cm Handle poll which changes the cm handle state from 'LOCKED' to 'ADVISED'. + * Find any failed (locked) cm handles and change state back to 'ADVISED'. */ @Scheduled(fixedDelayString = "${timers.locked-modules-sync.sleep-time-ms:300000}") - public void executeLockedCmHandlePoll() { - final List<YangModelCmHandle> lockedCmHandles = syncUtils.getModuleSyncFailedCmHandles(); - for (final YangModelCmHandle lockedCmHandle : lockedCmHandles) { - final CompositeState compositeState = lockedCmHandle.getCompositeState(); - final boolean isReadyForRetry = syncUtils.isReadyForRetry(compositeState); - if (isReadyForRetry) { - log.debug("Reset cm handle {} state to ADVISED to re-attempt module-sync", lockedCmHandle.getId()); - lcmEventsCmHandleStateHandler.updateCmHandleState(lockedCmHandle, CmHandleState.ADVISED); - } - } + public void resetPreviouslyFailedCmHandles() { + final List<YangModelCmHandle> failedCmHandles = syncUtils.getModuleSyncFailedCmHandles(); + moduleSyncTasks.resetFailedCmHandles(failedCmHandles); } - private void setCmHandleStateLocked(final YangModelCmHandle advisedCmHandle, - final CompositeState.LockReason lockReason) { - advisedCmHandle.getCompositeState().setLockReason(lockReason); - lcmEventsCmHandleStateHandler.updateCmHandleState(advisedCmHandle, CmHandleState.LOCKED); + private void preventBusyWait() { + // This method isn't really needed until CPS-1200 Performance Improvement: Watchdog Parallel execution + // but leaving here to minimize impacts on this class for that Jira + try { + TimeUnit.MILLISECONDS.sleep(PREVENT_CPU_BURN_WAIT_TIME_MILLIS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } } - private void updateModuleSyncSemaphoreMap(final String cmHandleId) { - moduleSyncSemaphores.replace(cmHandleId, MODEL_SYNC_DONE); + private void populateWorkQueueIfNeeded() { + if (moduleSyncWorkQueue.isEmpty()) { + final List<DataNode> advisedCmHandles = syncUtils.getAdvisedCmHandles(); + for (final DataNode advisedCmHandle : advisedCmHandles) { + if (!moduleSyncWorkQueue.offer(advisedCmHandle)) { + log.warn("Unable to add cm handle {} to the work queue", advisedCmHandle.getLeaves().get("id")); + } + } + } } - private boolean hasPushedIntoSemaphoreMap(final String cmHandleId) { - return moduleSyncSemaphores.putIfAbsent(cmHandleId, MODEL_SYNC_IN_PROGRESS) == null; + private Collection<DataNode> prepareNextBatch() { + final Collection<DataNode> nextBatchCandidates = new HashSet<>(MODULE_SYNC_BATCH_SIZE); + final Collection<DataNode> nextBatch = new HashSet<>(MODULE_SYNC_BATCH_SIZE); + moduleSyncWorkQueue.drainTo(nextBatchCandidates, MODULE_SYNC_BATCH_SIZE); + 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)); + if (alreadyAddedToInProgressMap) { + log.debug("module sync for {} already in progress by other instance", cmHandleId); + } else { + nextBatch.add(batchCandidate); + } + } + log.debug("nextBatch size : {}", nextBatch.size()); + return nextBatch; } + } |