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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2021-2024 Nordix Foundation.
   4  * ================================================================================
   5  * Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  * SPDX-License-Identifier: Apache-2.0
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.policy.clamp.acm.participant.intermediary.handler;
  24
  25 import io.micrometer.core.annotation.Timed;
  26 import lombok.RequiredArgsConstructor;
  27 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
  28 import org.onap.policy.clamp.models.acm.concepts.ParticipantState;
  29 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionDeploy;
  30 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionMigration;
  31 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionPrepare;
  32 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionStateChange;
  33 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantAckMessage;
  34 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregister;
  35 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregisterAck;
  36 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantMessage;
  37 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrime;
  38 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegister;
  39 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegisterAck;
  40 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatus;
  41 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatusReq;
  42 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
  43 import org.onap.policy.clamp.models.acm.messages.kafka.participant.PropertiesUpdate;
  44 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.DeployOrder;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47 import org.springframework.stereotype.Component;
  48
  49 /**
  50  * This class is responsible for managing the state of a participant.
  51  */
  52 @Component
  53 @RequiredArgsConstructor
  54 public class ParticipantHandler {
  55     private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantHandler.class);
  56
  57     private final AutomationCompositionHandler automationCompositionHandler;
  58     private final AcLockHandler acLockHandler;
  59     private final AcSubStateHandler acSubStateHandler;
  60     private final AcDefinitionHandler acDefinitionHandler;
  61     private final ParticipantMessagePublisher publisher;
  62     private final CacheProvider cacheProvider;
  63
  64     /**
  65      * Method which handles a participant health check event from clamp.
  66      *
  67      * @param participantStatusReqMsg participant participantStatusReq message
  68      */
  69     @Timed(value = "listener.participant_status_req", description = "PARTICIPANT_STATUS_REQ messages received")
  70     public void handleParticipantStatusReq(final ParticipantStatusReq participantStatusReqMsg) {
  71         sendHeartbeat();
  72     }
  73
  74     /**
  75      * Handle a automation composition update message.
  76      *
  77      * @param updateMsg the update message
  78      */
  79     @Timed(
  80             value = "listener.automation_composition_update",
  81             description = "AUTOMATION_COMPOSITION_UPDATE messages received")
  82     public void handleAutomationCompositionDeploy(AutomationCompositionDeploy updateMsg) {
  83         automationCompositionHandler.handleAutomationCompositionDeploy(updateMsg);
  84     }
  85
  86     /**
  87      * Handle a automation composition state change message.
  88      *
  89      * @param stateChangeMsg the state change message
  90      */
  91     @Timed(
  92             value = "listener.automation_composition_state_change",
  93             description = "AUTOMATION_COMPOSITION_STATE_CHANGE messages received")
  94     public void handleAutomationCompositionStateChange(AutomationCompositionStateChange stateChangeMsg) {
  95         if (DeployOrder.NONE.equals(stateChangeMsg.getDeployOrderedState())) {
  96             acLockHandler.handleAutomationCompositionStateChange(stateChangeMsg);
  97         } else {
  98             automationCompositionHandler.handleAutomationCompositionStateChange(stateChangeMsg);
  99         }
 100     }
 101
 102     /**
 103      * Handle a automation composition migration message.
 104      *
 105      * @param migrationMsg the migration message
 106      */
 107     @Timed(
 108             value = "listener.automation_composition_migration",
 109             description = "AUTOMATION_COMPOSITION_MIGRATION messages received")
 110     public void handleAutomationCompositionMigration(AutomationCompositionMigration migrationMsg) {
 111         if (Boolean.TRUE.equals(migrationMsg.getPrecheck())) {
 112             acSubStateHandler.handleAcMigrationPrecheck(migrationMsg);
 113         } else {
 114             automationCompositionHandler.handleAutomationCompositionMigration(migrationMsg);
 115         }
 116     }
 117
 118     /**
 119      * Handle a automation composition property update message.
 120      *
 121      * @param propertyUpdateMsg the property update message
 122      */
 123     @Timed(value = "listener.properties_update", description = "PROPERTIES_UPDATE message received")
 124     public void handleAcPropertyUpdate(PropertiesUpdate propertyUpdateMsg) {
 125         automationCompositionHandler.handleAcPropertyUpdate(propertyUpdateMsg);
 126     }
 127
 128     @Timed(value = "listener.prepare", description = "AUTOMATION_COMPOSITION_PREPARE message received")
 129     public void handleAutomationCompositionPrepare(AutomationCompositionPrepare acPrepareMsg) {
 130         acSubStateHandler.handleAcPrepare(acPrepareMsg);
 131     }
 132
 133     /**
 134      * Check if a participant message applies to this participant handler.
 135      *
 136      * @param participantMsg the message to check
 137      * @return true if it applies, false otherwise
 138      */
 139     public boolean appliesTo(ParticipantMessage participantMsg) {
 140         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 141     }
 142
 143     /**
 144      * Check if a participant message applies to this participant handler.
 145      *
 146      * @param participantMsg the message to check
 147      * @return true if it applies, false otherwise
 148      */
 149     public boolean appliesTo(ParticipantAckMessage participantMsg) {
 150         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 151     }
 152
 153     /**
 154      * Method to send ParticipantRegister message to automation composition runtime.
 155      */
 156     public void sendParticipantRegister() {
 157         var participantRegister = new ParticipantRegister();
 158         participantRegister.setParticipantId(cacheProvider.getParticipantId());
 159         participantRegister.setReplicaId(cacheProvider.getReplicaId());
 160         participantRegister.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 161
 162         publisher.sendParticipantRegister(participantRegister);
 163     }
 164
 165     /**
 166      * Handle a participantRegister Ack message.
 167      *
 168      * @param participantRegisterAckMsg the participantRegisterAck message
 169      */
 170     @Timed(value = "listener.participant_register_ack", description = "PARTICIPANT_REGISTER_ACK messages received")
 171     public void handleParticipantRegisterAck(ParticipantRegisterAck participantRegisterAckMsg) {
 172         LOGGER.debug("ParticipantRegisterAck message received as responseTo {}",
 173                 participantRegisterAckMsg.getResponseTo());
 174         cacheProvider.setRegistered(true);
 175         publisher.sendParticipantStatus(makeHeartbeat());
 176     }
 177
 178     /**
 179      * Method to send ParticipantDeregister message to automation composition runtime.
 180      */
 181     public void sendParticipantDeregister() {
 182         var participantDeregister = new ParticipantDeregister();
 183         participantDeregister.setParticipantId(cacheProvider.getParticipantId());
 184         participantDeregister.setReplicaId(cacheProvider.getReplicaId());
 185         publisher.sendParticipantDeregister(participantDeregister);
 186     }
 187
 188     /**
 189      * Handle a participantDeregister Ack message.
 190      *
 191      * @param participantDeregisterAckMsg the participantDeregisterAck message
 192      */
 193     @Timed(value = "listener.participant_deregister_ack", description = "PARTICIPANT_DEREGISTER_ACK messages received")
 194     public void handleParticipantDeregisterAck(ParticipantDeregisterAck participantDeregisterAckMsg) {
 195         LOGGER.debug("ParticipantDeregisterAck message received as responseTo {}",
 196                 participantDeregisterAckMsg.getResponseTo());
 197     }
 198
 199     /**
 200      * Handle a ParticipantPrime message.
 201      *
 202      * @param participantPrimeMsg the ParticipantPrime message
 203      */
 204     @Timed(value = "listener.participant_prime", description = "PARTICIPANT_PRIME messages received")
 205     public void handleParticipantPrime(ParticipantPrime participantPrimeMsg) {
 206         LOGGER.debug("ParticipantPrime message received for participantId {}", participantPrimeMsg.getParticipantId());
 207         acDefinitionHandler.handlePrime(participantPrimeMsg);
 208     }
 209
 210     /**
 211      * Handle a ParticipantSync message.
 212      *
 213      * @param participantSyncMsg the participantSync message
 214      */
 215     @Timed(value = "listener.participant_sync_msg", description = "PARTICIPANT_SYNC messages received")
 216     public void handleParticipantSync(ParticipantSync participantSyncMsg) {
 217         if (participantSyncMsg.getExcludeReplicas().contains(cacheProvider.getReplicaId())) {
 218             LOGGER.debug("Ignore ParticipantSync message {}", participantSyncMsg.getMessageId());
 219             return;
 220         }
 221         LOGGER.debug("ParticipantSync message received for participantId {}", participantSyncMsg.getParticipantId());
 222         acDefinitionHandler.handleParticipantSync(participantSyncMsg);
 223     }
 224
 225     /**
 226      * Dispatch a heartbeat for this participant.
 227      */
 228     public void sendHeartbeat() {
 229         if (publisher.isActive()) {
 230             if (!cacheProvider.isRegistered()) {
 231                 sendParticipantRegister();
 232             } else {
 233                 publisher.sendParticipantStatus(makeHeartbeat());
 234             }
 235         }
 236     }
 237
 238     /**
 239      * Method to send heartbeat to automation composition runtime.
 240      */
 241     private ParticipantStatus makeHeartbeat() {
 242         var heartbeat = new ParticipantStatus();
 243         heartbeat.setParticipantId(cacheProvider.getParticipantId());
 244         heartbeat.setReplicaId(cacheProvider.getReplicaId());
 245         heartbeat.setState(ParticipantState.ON_LINE);
 246         heartbeat.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 247
 248         return heartbeat;
 249     }
 250 }