gerrit.onap Code Review - policy/clamp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2023 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.clamp.acm.runtime.supervision;
  22
  23 import io.micrometer.core.annotation.Timed;
  24 import java.util.ArrayList;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.UUID;
  29 import lombok.AllArgsConstructor;
  30 import org.apache.commons.collections4.MapUtils;
  31 import org.onap.policy.clamp.acm.runtime.supervision.comm.ParticipantDeregisterAckPublisher;
  32 import org.onap.policy.clamp.acm.runtime.supervision.comm.ParticipantRegisterAckPublisher;
  33 import org.onap.policy.clamp.acm.runtime.supervision.comm.ParticipantRestartPublisher;
  34 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
  35 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
  36 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionDefinition;
  37 import org.onap.policy.clamp.models.acm.concepts.Participant;
  38 import org.onap.policy.clamp.models.acm.concepts.ParticipantState;
  39 import org.onap.policy.clamp.models.acm.concepts.ParticipantSupportedElementType;
  40 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
  41 import org.onap.policy.clamp.models.acm.messages.dmaap.participant.ParticipantDeregister;
  42 import org.onap.policy.clamp.models.acm.messages.dmaap.participant.ParticipantRegister;
  43 import org.onap.policy.clamp.models.acm.messages.dmaap.participant.ParticipantStatus;
  44 import org.onap.policy.clamp.models.acm.persistence.provider.AcDefinitionProvider;
  45 import org.onap.policy.clamp.models.acm.persistence.provider.AutomationCompositionProvider;
  46 import org.onap.policy.clamp.models.acm.persistence.provider.ParticipantProvider;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49 import org.springframework.stereotype.Component;
  50
  51 /**
  52  * This class handles supervision of participant status.
  53  */
  54 @Component
  55 @AllArgsConstructor
  56 public class SupervisionParticipantHandler {
  57     private static final Logger LOGGER = LoggerFactory.getLogger(SupervisionParticipantHandler.class);
  58
  59     private final ParticipantProvider participantProvider;
  60     private final ParticipantRegisterAckPublisher participantRegisterAckPublisher;
  61     private final ParticipantDeregisterAckPublisher participantDeregisterAckPublisher;
  62     private final AutomationCompositionProvider automationCompositionProvider;
  63     private final AcDefinitionProvider acDefinitionProvider;
  64     private final ParticipantRestartPublisher participantRestartPublisher;
  65
  66     /**
  67      * Handle a ParticipantRegister message from a participant.
  68      *
  69      * @param participantRegisterMsg the ParticipantRegister message received from a participant
  70      */
  71     @MessageIntercept
  72     @Timed(value = "listener.participant_register", description = "PARTICIPANT_REGISTER messages received")
  73     public void handleParticipantMessage(ParticipantRegister participantRegisterMsg) {
  74         LOGGER.debug("Participant Register received {}", participantRegisterMsg);
  75         var participantOpt = participantProvider.findParticipant(participantRegisterMsg.getParticipantId());
  76
  77         if (participantOpt.isPresent()) {
  78             var participant = participantOpt.get();
  79             checkOnline(participant);
  80             handleRestart(participant.getParticipantId());
  81         } else {
  82             var participant = createParticipant(participantRegisterMsg.getParticipantId(),
  83                     listToMap(participantRegisterMsg.getParticipantSupportedElementType()));
  84             participantProvider.saveParticipant(participant);
  85
  86         }
  87
  88         participantRegisterAckPublisher.send(participantRegisterMsg.getMessageId(),
  89                 participantRegisterMsg.getParticipantId());
  90     }
  91
  92     /**
  93      * Handle a ParticipantDeregister message from a participant.
  94      *
  95      * @param participantDeregisterMsg the ParticipantDeregister message received from a participant
  96      */
  97     @MessageIntercept
  98     @Timed(value = "listener.participant_deregister", description = "PARTICIPANT_DEREGISTER messages received")
  99     public void handleParticipantMessage(ParticipantDeregister participantDeregisterMsg) {
 100         LOGGER.debug("Participant Deregister received {}", participantDeregisterMsg);
 101         var participantOpt = participantProvider.findParticipant(participantDeregisterMsg.getParticipantId());
 102
 103         if (participantOpt.isPresent()) {
 104             var participant = participantOpt.get();
 105             participant.setParticipantState(ParticipantState.OFF_LINE);
 106             participantProvider.updateParticipant(participant);
 107         }
 108
 109         participantDeregisterAckPublisher.send(participantDeregisterMsg.getMessageId());
 110     }
 111
 112     /**
 113      * Handle a ParticipantStatus message from a participant.
 114      *
 115      * @param participantStatusMsg the ParticipantStatus message received from a participant
 116      */
 117     @MessageIntercept
 118     @Timed(value = "listener.participant_status", description = "PARTICIPANT_STATUS messages received")
 119     public void handleParticipantMessage(ParticipantStatus participantStatusMsg) {
 120         LOGGER.debug("Participant Status received {}", participantStatusMsg);
 121
 122         var participantOpt = participantProvider.findParticipant(participantStatusMsg.getParticipantId());
 123         if (participantOpt.isEmpty()) {
 124             var participant = createParticipant(participantStatusMsg.getParticipantId(),
 125                     listToMap(participantStatusMsg.getParticipantSupportedElementType()));
 126             participantProvider.saveParticipant(participant);
 127         } else {
 128             checkOnline(participantOpt.get());
 129         }
 130         if (!participantStatusMsg.getAutomationCompositionInfoList().isEmpty()) {
 131             automationCompositionProvider.upgradeStates(participantStatusMsg.getAutomationCompositionInfoList());
 132         }
 133     }
 134
 135     private void checkOnline(Participant participant) {
 136         if (ParticipantState.OFF_LINE.equals(participant.getParticipantState())) {
 137             participant.setParticipantState(ParticipantState.ON_LINE);
 138             participantProvider.saveParticipant(participant);
 139         }
 140     }
 141
 142     private void handleRestart(UUID participantId) {
 143         var compositionIds = participantProvider.getCompositionIds(participantId);
 144         for (var compositionId : compositionIds) {
 145             var acDefinition = acDefinitionProvider.getAcDefinition(compositionId);
 146             LOGGER.debug("Scan Composition {} for restart", acDefinition.getCompositionId());
 147             handleRestart(participantId, acDefinition);
 148         }
 149     }
 150
 151     private void handleRestart(UUID participantId, AutomationCompositionDefinition acDefinition) {
 152         if (AcTypeState.COMMISSIONED.equals(acDefinition.getState())) {
 153             LOGGER.debug("Composition {} COMMISSIONED", acDefinition.getCompositionId());
 154             return;
 155         }
 156         LOGGER.debug("Composition to be send in Restart message {}", acDefinition.getCompositionId());
 157         for (var elementState : acDefinition.getElementStateMap().values()) {
 158             if (participantId.equals(elementState.getParticipantId())) {
 159                 elementState.setRestarting(true);
 160             }
 161         }
 162         var automationCompositionList =
 163                 automationCompositionProvider.getAcInstancesByCompositionId(acDefinition.getCompositionId());
 164         List<AutomationComposition> automationCompositions = new ArrayList<>();
 165         for (var automationComposition : automationCompositionList) {
 166             if (isAcToBeRestarted(participantId, automationComposition)) {
 167                 automationCompositions.add(automationComposition);
 168             }
 169         }
 170         // expected final state
 171         if (StateChangeResult.TIMEOUT.equals(acDefinition.getStateChangeResult())) {
 172             acDefinition.setStateChangeResult(StateChangeResult.NO_ERROR);
 173         }
 174         acDefinition.setRestarting(true);
 175         acDefinitionProvider.updateAcDefinition(acDefinition);
 176         participantRestartPublisher.send(participantId, acDefinition, automationCompositions);
 177     }
 178
 179     private boolean isAcToBeRestarted(UUID participantId, AutomationComposition automationComposition) {
 180         boolean toAdd = false;
 181         for (var element : automationComposition.getElements().values()) {
 182             if (participantId.equals(element.getParticipantId())) {
 183                 element.setRestarting(true);
 184                 toAdd = true;
 185             }
 186         }
 187         if (toAdd) {
 188             automationComposition.setRestarting(true);
 189             // expected final state
 190             if (StateChangeResult.TIMEOUT.equals(automationComposition.getStateChangeResult())) {
 191                 automationComposition.setStateChangeResult(StateChangeResult.NO_ERROR);
 192             }
 193             automationCompositionProvider.updateAutomationComposition(automationComposition);
 194         }
 195         return toAdd;
 196     }
 197
 198     private Participant createParticipant(UUID participantId,
 199             Map<UUID, ParticipantSupportedElementType> participantSupportedElementType) {
 200         var participant = new Participant();
 201         participant.setParticipantId(participantId);
 202         participant.setParticipantSupportedElementTypes(participantSupportedElementType);
 203         participant.setParticipantState(ParticipantState.ON_LINE);
 204         return participant;
 205     }
 206
 207     private Map<UUID, ParticipantSupportedElementType> listToMap(List<ParticipantSupportedElementType> elementList) {
 208         Map<UUID, ParticipantSupportedElementType> map = new HashMap<>();
 209         MapUtils.populateMap(map, elementList, ParticipantSupportedElementType::getId);
 210         return map;
 211     }
 212 }