diff options
Diffstat (limited to 'runtime-acm/src/main')
4 files changed, 111 insertions, 102 deletions
diff --git a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/instantiation/AutomationCompositionInstantiationProvider.java b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/instantiation/AutomationCompositionInstantiationProvider.java index 653d97338..51b7b714c 100644 --- a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/instantiation/AutomationCompositionInstantiationProvider.java +++ b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/instantiation/AutomationCompositionInstantiationProvider.java @@ -23,7 +23,6 @@ package org.onap.policy.clamp.acm.runtime.instantiation; import jakarta.validation.Valid; import jakarta.ws.rs.core.Response.Status; -import java.util.ArrayList; import java.util.List; import java.util.UUID; import java.util.stream.Collectors; @@ -68,6 +67,7 @@ import org.springframework.transaction.annotation.Transactional; @RequiredArgsConstructor public class AutomationCompositionInstantiationProvider { private static final String DO_NOT_MATCH = " do not match with "; + private static final String ELEMENT_ID_NOT_PRESENT = "Element id not present "; private static final Logger LOGGER = LoggerFactory.getLogger(AutomationCompositionInstantiationProvider.class); @@ -190,7 +190,7 @@ public class AutomationCompositionInstantiationProvider { var elementId = element.getKey(); var dbAcElement = acToBeUpdated.getElements().get(elementId); if (dbAcElement == null) { - throw new PfModelRuntimeException(Status.BAD_REQUEST, "Element id not present " + elementId); + throw new PfModelRuntimeException(Status.BAD_REQUEST, ELEMENT_ID_NOT_PRESENT + elementId); } AcmUtils.recursiveMerge(dbAcElement.getProperties(), element.getValue().getProperties()); } @@ -221,7 +221,7 @@ public class AutomationCompositionInstantiationProvider { // Add additional elements if present for migration if (dbAcElement == null) { LOGGER.info("New Ac element {} added in Migration", elementId); - acToBeUpdated.getElements().put(elementId, automationComposition.getElements().get(elementId)); + acToBeUpdated.getElements().put(elementId, element.getValue()); } else { AcmUtils.recursiveMerge(dbAcElement.getProperties(), element.getValue().getProperties()); var newDefinition = element.getValue().getDefinition().asConceptKey(); @@ -231,30 +231,27 @@ public class AutomationCompositionInstantiationProvider { } } // Remove element which is not present in the new Ac instance - List<UUID> elementsRemoved = new ArrayList<>(); - for (var dbElement : acToBeUpdated.getElements().entrySet()) { - var dbElementId = dbElement.getKey(); - if (automationComposition.getElements().get(dbElementId) == null) { - LOGGER.info("Element with id {} is removed in Migration", dbElementId); - elementsRemoved.add(dbElementId); - automationCompositionProvider.deleteAutomationCompositionElement(dbElementId); - } - } + var elementsRemoved = getElementRemoved(acToBeUpdated, automationComposition); elementsRemoved.forEach(uuid -> acToBeUpdated.getElements().remove(uuid)); var validationResult = - validateAutomationComposition(acToBeUpdated, automationComposition.getCompositionTargetId()); + validateAutomationComposition(acToBeUpdated, automationComposition.getCompositionTargetId()); if (!validationResult.isValid()) { throw new PfModelRuntimeException(Status.BAD_REQUEST, validationResult.getResult()); } acToBeUpdated.setCompositionTargetId(automationComposition.getCompositionTargetId()); - + var acDefinition = acDefinitionProvider.getAcDefinition(automationComposition.getCompositionTargetId()); // Publish migrate event to the participants - supervisionAcHandler.migrate(acToBeUpdated); + supervisionAcHandler.migrate(acToBeUpdated, acDefinition.getServiceTemplate()); - automationComposition = automationCompositionProvider.updateAutomationComposition(acToBeUpdated); + var ac = automationCompositionProvider.updateAutomationComposition(acToBeUpdated); elementsRemoved.forEach(automationCompositionProvider::deleteAutomationCompositionElement); - return createInstantiationResponse(automationComposition); + return createInstantiationResponse(ac); + } + + private List<UUID> getElementRemoved(AutomationComposition acFromDb, AutomationComposition acFromMigration) { + return acFromDb.getElements().keySet().stream() + .filter(id -> acFromMigration.getElements().get(id) == null).toList(); } void checkCompatibility(PfConceptKey newDefinition, PfConceptKey dbElementDefinition, @@ -283,7 +280,7 @@ public class AutomationCompositionInstantiationProvider { // Add additional elements if present for migration if (copyElement == null) { LOGGER.info("New Ac element {} added in Migration", elementId); - copyAc.getElements().put(elementId, automationComposition.getElements().get(elementId)); + copyAc.getElements().put(elementId, element.getValue()); } else { AcmUtils.recursiveMerge(copyElement.getProperties(), element.getValue().getProperties()); var newDefinition = element.getValue().getDefinition().asConceptKey(); @@ -293,15 +290,7 @@ public class AutomationCompositionInstantiationProvider { } } // Remove element which is not present in the new Ac instance - List<UUID> elementsRemoved = new ArrayList<>(); - for (var dbElement : copyAc.getElements().entrySet()) { - var dbElementId = dbElement.getKey(); - if (automationComposition.getElements().get(dbElementId) == null) { - LOGGER.info("Element with id {} is removed in Migration", dbElementId); - elementsRemoved.add(dbElementId); - } - } - + var elementsRemoved = getElementRemoved(copyAc, automationComposition); elementsRemoved.forEach(uuid -> copyAc.getElements().remove(uuid)); var validationResult = @@ -317,6 +306,8 @@ public class AutomationCompositionInstantiationProvider { AcmUtils.setCascadedState(acToBeUpdated, DeployState.DEPLOYED, LockState.LOCKED, SubState.MIGRATION_PRECHECKING); acToBeUpdated.setStateChangeResult(StateChangeResult.NO_ERROR); + // excluding removed element in MIGRATION_PRECHECKING + elementsRemoved.forEach(uuid -> acToBeUpdated.getElements().get(uuid).setSubState(SubState.NONE)); return createInstantiationResponse(automationCompositionProvider.updateAutomationComposition(acToBeUpdated)); } diff --git a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionAcHandler.java b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionAcHandler.java index a8d26871a..66f035e5e 100644 --- a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionAcHandler.java +++ b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionAcHandler.java @@ -48,6 +48,7 @@ import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCom import org.onap.policy.clamp.models.acm.persistence.provider.AcDefinitionProvider; import org.onap.policy.clamp.models.acm.persistence.provider.AutomationCompositionProvider; import org.onap.policy.clamp.models.acm.utils.AcmUtils; +import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; @@ -287,7 +288,7 @@ public class SupervisionAcHandler { var updated = updateState(automationComposition, automationCompositionAckMessage.getAutomationCompositionResultMap().entrySet(), - automationCompositionAckMessage.getStateChangeResult()); + automationCompositionAckMessage.getStateChangeResult(), automationCompositionAckMessage.getStage()); if (updated) { automationComposition = automationCompositionProvider.updateAcState(automationComposition); var acDefinition = acDefinitionProvider.getAcDefinition(automationComposition.getCompositionId()); @@ -297,7 +298,7 @@ public class SupervisionAcHandler { private boolean updateState(AutomationComposition automationComposition, Set<Map.Entry<UUID, AcElementDeployAck>> automationCompositionResultSet, - StateChangeResult stateChangeResult) { + StateChangeResult stateChangeResult, Integer stage) { var updated = false; boolean inProgress = !StateChangeResult.FAILED.equals(automationComposition.getStateChangeResult()); if (inProgress && !stateChangeResult.equals(automationComposition.getStateChangeResult())) { @@ -315,6 +316,7 @@ public class SupervisionAcHandler { element.setSubState(SubState.NONE); element.setDeployState(acElementAck.getValue().getDeployState()); element.setLockState(acElementAck.getValue().getLockState()); + element.setStage(stage); element.setRestarting(null); automationCompositionProvider.updateAutomationCompositionElement(element); } @@ -336,11 +338,14 @@ public class SupervisionAcHandler { * Handle Migration of an AutomationComposition instance to other ACM Definition. * * @param automationComposition the AutomationComposition + * @param serviceTemplate the ServiceTemplate */ - public void migrate(AutomationComposition automationComposition) { + public void migrate(AutomationComposition automationComposition, ToscaServiceTemplate serviceTemplate) { AcmUtils.setCascadedState(automationComposition, DeployState.MIGRATING, LockState.LOCKED); + var stage = ParticipantUtils.getFirstStage(automationComposition, serviceTemplate); automationComposition.setStateChangeResult(StateChangeResult.NO_ERROR); - executor.execute(() -> acCompositionMigrationPublisher.send(automationComposition)); + automationComposition.setPhase(stage); + executor.execute(() -> acCompositionMigrationPublisher.send(automationComposition, stage)); } /** @@ -349,6 +354,6 @@ public class SupervisionAcHandler { * @param automationComposition the AutomationComposition */ public void migratePrecheck(AutomationComposition automationComposition) { - executor.execute(() -> acCompositionMigrationPublisher.send(automationComposition)); + executor.execute(() -> acCompositionMigrationPublisher.send(automationComposition, 0)); } } diff --git a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionScanner.java b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionScanner.java index e723d2c0b..31cc8babe 100644 --- a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionScanner.java +++ b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/SupervisionScanner.java @@ -22,10 +22,12 @@ package org.onap.policy.clamp.acm.runtime.supervision; +import java.util.Comparator; import java.util.HashMap; import java.util.UUID; import org.onap.policy.clamp.acm.runtime.main.parameters.AcRuntimeParameterGroup; import org.onap.policy.clamp.acm.runtime.supervision.comm.AutomationCompositionDeployPublisher; +import org.onap.policy.clamp.acm.runtime.supervision.comm.AutomationCompositionMigrationPublisher; import org.onap.policy.clamp.acm.runtime.supervision.comm.AutomationCompositionStateChangePublisher; import org.onap.policy.clamp.acm.runtime.supervision.comm.ParticipantSyncPublisher; import org.onap.policy.clamp.models.acm.concepts.AcTypeState; @@ -58,6 +60,7 @@ public class SupervisionScanner { private final AutomationCompositionStateChangePublisher automationCompositionStateChangePublisher; private final AutomationCompositionDeployPublisher automationCompositionDeployPublisher; private final ParticipantSyncPublisher participantSyncPublisher; + private final AutomationCompositionMigrationPublisher automationCompositionMigrationPublisher; /** * Constructor for instantiating SupervisionScanner. @@ -73,12 +76,14 @@ public class SupervisionScanner { final AutomationCompositionStateChangePublisher automationCompositionStateChangePublisher, final AutomationCompositionDeployPublisher automationCompositionDeployPublisher, final ParticipantSyncPublisher participantSyncPublisher, + final AutomationCompositionMigrationPublisher automationCompositionMigrationPublisher, final AcRuntimeParameterGroup acRuntimeParameterGroup) { this.automationCompositionProvider = automationCompositionProvider; this.acDefinitionProvider = acDefinitionProvider; this.automationCompositionStateChangePublisher = automationCompositionStateChangePublisher; this.automationCompositionDeployPublisher = automationCompositionDeployPublisher; this.participantSyncPublisher = participantSyncPublisher; + this.automationCompositionMigrationPublisher = automationCompositionMigrationPublisher; this.maxStatusWaitMs = acRuntimeParameterGroup.getParticipantParameters().getMaxStatusWaitMs(); } @@ -96,11 +101,9 @@ public class SupervisionScanner { var acList = automationCompositionProvider.getAcInstancesInTransition(); HashMap<UUID, AutomationCompositionDefinition> acDefinitionMap = new HashMap<>(); for (var automationComposition : acList) { - var acDefinition = acDefinitionMap.get(automationComposition.getCompositionId()); - if (acDefinition == null) { - acDefinition = acDefinitionProvider.getAcDefinition(automationComposition.getCompositionId()); - acDefinitionMap.put(acDefinition.getCompositionId(), acDefinition); - } + var compositionId = automationComposition.getCompositionTargetId() != null + ? automationComposition.getCompositionTargetId() : automationComposition.getCompositionId(); + var acDefinition = acDefinitionMap.computeIfAbsent(compositionId, acDefinitionProvider::getAcDefinition); scanAutomationComposition(automationComposition, acDefinition.getServiceTemplate()); } LOGGER.debug("Automation composition scan complete . . ."); @@ -118,6 +121,7 @@ public class SupervisionScanner { for (var element : acDefinition.getElementStateMap().values()) { if (!finalState.equals(element.getState())) { completed = false; + break; } } if (completed) { @@ -141,6 +145,17 @@ public class SupervisionScanner { return; } + if (DeployState.UPDATING.equals(automationComposition.getDeployState()) + || DeployState.MIGRATING.equals(automationComposition.getDeployState()) + || SubState.MIGRATION_PRECHECKING.equals(automationComposition.getSubState())) { + + scanMigrate(automationComposition, serviceTemplate); + } else { + scanDeploy(automationComposition, serviceTemplate); + } + } + + private void scanDeploy(final AutomationComposition automationComposition, ToscaServiceTemplate serviceTemplate) { var completed = true; var minSpNotCompleted = 1000; // min startPhase not completed var maxSpNotCompleted = 0; // max startPhase not completed @@ -169,23 +184,51 @@ public class SupervisionScanner { LOGGER.debug("automation composition scan: transition state {} {} not completed", automationComposition.getDeployState(), automationComposition.getLockState()); - if (DeployState.UPDATING.equals(automationComposition.getDeployState()) - || DeployState.MIGRATING.equals(automationComposition.getDeployState()) - || !SubState.NONE.equals(automationComposition.getSubState())) { - // UPDATING do not need phases - handleTimeoutUpdate(automationComposition); - return; - } - var isForward = AcmUtils.isForward(automationComposition.getDeployState(), automationComposition.getLockState()); var nextSpNotCompleted = isForward ? minSpNotCompleted : maxSpNotCompleted; if (nextSpNotCompleted != automationComposition.getPhase()) { - sendAutomationCompositionMsg(automationComposition, serviceTemplate, nextSpNotCompleted, false); + sendAutomationCompositionMsg(automationComposition, serviceTemplate, nextSpNotCompleted); + } else { + handleTimeout(automationComposition); + } + } + } + + private void scanMigrate(final AutomationComposition automationComposition, ToscaServiceTemplate serviceTemplate) { + var completed = true; + var minStageNotCompleted = 1000; // min stage not completed + for (var element : automationComposition.getElements().values()) { + if (AcmUtils.isInTransitionalState(element.getDeployState(), element.getLockState(), + element.getSubState())) { + var toscaNodeTemplate = serviceTemplate.getToscaTopologyTemplate().getNodeTemplates() + .get(element.getDefinition().getName()); + var stageSet = ParticipantUtils.findStageSet(toscaNodeTemplate.getProperties()); + var minStage = stageSet.stream().min(Comparator.comparing(Integer::valueOf)).orElse(0); + int stage = element.getStage() != null ? element.getStage() : minStage; + minStageNotCompleted = Math.min(minStageNotCompleted, stage); + completed = false; + } + } + + if (completed) { + LOGGER.debug("automation composition scan: transition state {} {} ", automationComposition.getDeployState(), + automationComposition.getLockState()); + + complete(automationComposition, serviceTemplate); + } else { + LOGGER.debug("automation composition scan: transition from state {} to {} not completed", + automationComposition.getDeployState(), automationComposition.getLockState()); + + if (DeployState.MIGRATING.equals(automationComposition.getDeployState()) + && minStageNotCompleted != automationComposition.getPhase()) { + savePahese(automationComposition, minStageNotCompleted); + LOGGER.debug("retry message AutomationCompositionMigration"); + automationCompositionMigrationPublisher.send(automationComposition, minStageNotCompleted); } else { - handleTimeoutWithPhase(automationComposition, serviceTemplate); + handleTimeout(automationComposition); } } } @@ -230,69 +273,37 @@ public class SupervisionScanner { } } - private void handleTimeoutUpdate(AutomationComposition automationComposition) { + private void handleTimeout(AutomationComposition automationComposition) { if (StateChangeResult.TIMEOUT.equals(automationComposition.getStateChangeResult())) { LOGGER.debug("The ac instance is in timeout {}", automationComposition.getInstanceId()); return; } var now = TimestampHelper.nowEpochMilli(); var lastMsg = TimestampHelper.toEpochMilli(automationComposition.getLastMsg()); - for (var element : automationComposition.getElements().values()) { - if (!AcmUtils.isInTransitionalState( - element.getDeployState(), element.getLockState(), element.getSubState())) { - continue; - } - if ((now - lastMsg) > maxStatusWaitMs) { - LOGGER.debug("Report timeout for the ac instance {}", automationComposition.getInstanceId()); - automationComposition.setStateChangeResult(StateChangeResult.TIMEOUT); - automationCompositionProvider.updateAutomationComposition(automationComposition); - break; - } + if ((now - lastMsg) > maxStatusWaitMs) { + LOGGER.debug("Report timeout for the ac instance {}", automationComposition.getInstanceId()); + automationComposition.setStateChangeResult(StateChangeResult.TIMEOUT); + automationCompositionProvider.updateAcState(automationComposition); } } - private void handleTimeoutWithPhase(AutomationComposition automationComposition, - ToscaServiceTemplate serviceTemplate) { - if (StateChangeResult.TIMEOUT.equals(automationComposition.getStateChangeResult())) { - LOGGER.debug("The ac instance is in timeout {}", automationComposition.getInstanceId()); - return; - } - int currentPhase = automationComposition.getPhase(); - var now = TimestampHelper.nowEpochMilli(); - var lastMsg = TimestampHelper.toEpochMilli(automationComposition.getLastMsg()); - for (var element : automationComposition.getElements().values()) { - if (!AcmUtils.isInTransitionalState( - element.getDeployState(), element.getLockState(), element.getSubState())) { - continue; - } - var toscaNodeTemplate = serviceTemplate.getToscaTopologyTemplate().getNodeTemplates() - .get(element.getDefinition().getName()); - int startPhase = ParticipantUtils.findStartPhase(toscaNodeTemplate.getProperties()); - if (currentPhase != startPhase) { - continue; - } - if ((now - lastMsg) > maxStatusWaitMs) { - LOGGER.debug("Report timeout for the ac instance {}", automationComposition.getInstanceId()); - automationComposition.setStateChangeResult(StateChangeResult.TIMEOUT); - automationCompositionProvider.updateAutomationComposition(automationComposition); - break; - } - } + private void savePahese(AutomationComposition automationComposition, int startPhase) { + automationComposition.setLastMsg(TimestampHelper.now()); + automationComposition.setPhase(startPhase); + automationCompositionProvider.updateAcState(automationComposition); } private void sendAutomationCompositionMsg(AutomationComposition automationComposition, - ToscaServiceTemplate serviceTemplate, int startPhase, boolean firstStartPhase) { - automationComposition.setLastMsg(TimestampHelper.now()); - automationComposition.setPhase(startPhase); - automationCompositionProvider.updateAutomationComposition(automationComposition); + ToscaServiceTemplate serviceTemplate, int startPhase) { + savePahese(automationComposition, startPhase); if (DeployState.DEPLOYING.equals(automationComposition.getDeployState())) { - LOGGER.debug("retry message AutomationCompositionUpdate"); + LOGGER.debug("retry message AutomationCompositionDeploy"); automationCompositionDeployPublisher.send(automationComposition, serviceTemplate, startPhase, - firstStartPhase); + false); } else { LOGGER.debug("retry message AutomationCompositionStateChange"); - automationCompositionStateChangePublisher.send(automationComposition, startPhase, firstStartPhase); + automationCompositionStateChangePublisher.send(automationComposition, startPhase, false); } } } diff --git a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/comm/AutomationCompositionMigrationPublisher.java b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/comm/AutomationCompositionMigrationPublisher.java index b961eab93..572f7b1fe 100644 --- a/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/comm/AutomationCompositionMigrationPublisher.java +++ b/runtime-acm/src/main/java/org/onap/policy/clamp/acm/runtime/supervision/comm/AutomationCompositionMigrationPublisher.java @@ -36,20 +36,22 @@ public class AutomationCompositionMigrationPublisher * Send AutomationCompositionMigration message to Participant. * * @param automationComposition the AutomationComposition + * @param stage the stage to execute */ @Timed( value = "publisher.automation_composition_migration", description = "AUTOMATION_COMPOSITION_MIGRATION messages published") - public void send(AutomationComposition automationComposition) { - var acsc = new AutomationCompositionMigration(); - acsc.setPrecheck(Boolean.TRUE.equals(automationComposition.getPrecheck())); - acsc.setCompositionId(automationComposition.getCompositionId()); - acsc.setAutomationCompositionId(automationComposition.getInstanceId()); - acsc.setMessageId(UUID.randomUUID()); - acsc.setCompositionTargetId(automationComposition.getCompositionTargetId()); - acsc.setParticipantUpdatesList( + public void send(AutomationComposition automationComposition, int stage) { + var acMigration = new AutomationCompositionMigration(); + acMigration.setPrecheck(Boolean.TRUE.equals(automationComposition.getPrecheck())); + acMigration.setCompositionId(automationComposition.getCompositionId()); + acMigration.setAutomationCompositionId(automationComposition.getInstanceId()); + acMigration.setMessageId(UUID.randomUUID()); + acMigration.setCompositionTargetId(automationComposition.getCompositionTargetId()); + acMigration.setStage(stage); + acMigration.setParticipantUpdatesList( AcmUtils.createParticipantDeployList(automationComposition, DeployOrder.MIGRATE)); - super.send(acsc); + super.send(acMigration); } } |