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.AutomationCompositionStateChange;
  32 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantAckMessage;
  33 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregister;
  34 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregisterAck;
  35 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantMessage;
  36 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrime;
  37 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegister;
  38 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegisterAck;
  39 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRestart;
  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 AcDefinitionHandler acDefinitionHandler;
  60     private final ParticipantMessagePublisher publisher;
  61     private final CacheProvider cacheProvider;
  62
  63     /**
  64      * Method which handles a participant health check event from clamp.
  65      *
  66      * @param participantStatusReqMsg participant participantStatusReq message
  67      */
  68     @Timed(value = "listener.participant_status_req", description = "PARTICIPANT_STATUS_REQ messages received")
  69     public void handleParticipantStatusReq(final ParticipantStatusReq participantStatusReqMsg) {
  70         sendHeartbeat();
  71     }
  72
  73     /**
  74      * Handle a automation composition update message.
  75      *
  76      * @param updateMsg the update message
  77      */
  78     @Timed(
  79             value = "listener.automation_composition_update",
  80             description = "AUTOMATION_COMPOSITION_UPDATE messages received")
  81     public void handleAutomationCompositionDeploy(AutomationCompositionDeploy updateMsg) {
  82         automationCompositionHandler.handleAutomationCompositionDeploy(updateMsg);
  83     }
  84
  85     /**
  86      * Handle a automation composition state change message.
  87      *
  88      * @param stateChangeMsg the state change message
  89      */
  90     @Timed(
  91             value = "listener.automation_composition_state_change",
  92             description = "AUTOMATION_COMPOSITION_STATE_CHANGE messages received")
  93     public void handleAutomationCompositionStateChange(AutomationCompositionStateChange stateChangeMsg) {
  94         if (DeployOrder.NONE.equals(stateChangeMsg.getDeployOrderedState())) {
  95             acLockHandler.handleAutomationCompositionStateChange(stateChangeMsg);
  96         } else {
  97             automationCompositionHandler.handleAutomationCompositionStateChange(stateChangeMsg);
  98         }
  99     }
 100
 101     /**
 102      * Handle a automation composition migration message.
 103      *
 104      * @param migrationMsg the migration message
 105      */
 106     @Timed(
 107             value = "listener.automation_composition_migration",
 108             description = "AUTOMATION_COMPOSITION_MIGRATION messages received")
 109     public void handleAutomationCompositionMigration(AutomationCompositionMigration migrationMsg) {
 110         automationCompositionHandler.handleAutomationCompositionMigration(migrationMsg);
 111     }
 112
 113     /**
 114      * Handle a automation composition property update message.
 115      *
 116      * @param propertyUpdateMsg the property update message
 117      */
 118     @Timed(value = "listener.properties_update", description = "PROPERTIES_UPDATE message received")
 119     public void handleAcPropertyUpdate(PropertiesUpdate propertyUpdateMsg) {
 120         automationCompositionHandler.handleAcPropertyUpdate(propertyUpdateMsg);
 121     }
 122
 123     /**
 124      * Check if a participant message applies to this participant handler.
 125      *
 126      * @param participantMsg the message to check
 127      * @return true if it applies, false otherwise
 128      */
 129     public boolean appliesTo(ParticipantMessage participantMsg) {
 130         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 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(ParticipantAckMessage participantMsg) {
 140         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 141     }
 142
 143     /**
 144      * Method to send ParticipantRegister message to automation composition runtime.
 145      */
 146     public void sendParticipantRegister() {
 147         var participantRegister = new ParticipantRegister();
 148         participantRegister.setParticipantId(cacheProvider.getParticipantId());
 149         participantRegister.setReplicaId(cacheProvider.getReplicaId());
 150         participantRegister.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 151
 152         publisher.sendParticipantRegister(participantRegister);
 153     }
 154
 155     /**
 156      * Handle a participantRegister Ack message.
 157      *
 158      * @param participantRegisterAckMsg the participantRegisterAck message
 159      */
 160     @Timed(value = "listener.participant_register_ack", description = "PARTICIPANT_REGISTER_ACK messages received")
 161     public void handleParticipantRegisterAck(ParticipantRegisterAck participantRegisterAckMsg) {
 162         LOGGER.debug("ParticipantRegisterAck message received as responseTo {}",
 163                 participantRegisterAckMsg.getResponseTo());
 164         cacheProvider.setRegistered(true);
 165         publisher.sendParticipantStatus(makeHeartbeat());
 166     }
 167
 168     /**
 169      * Method to send ParticipantDeregister message to automation composition runtime.
 170      */
 171     public void sendParticipantDeregister() {
 172         var participantDeregister = new ParticipantDeregister();
 173         participantDeregister.setParticipantId(cacheProvider.getParticipantId());
 174         participantDeregister.setReplicaId(cacheProvider.getReplicaId());
 175         publisher.sendParticipantDeregister(participantDeregister);
 176     }
 177
 178     /**
 179      * Handle a participantDeregister Ack message.
 180      *
 181      * @param participantDeregisterAckMsg the participantDeregisterAck message
 182      */
 183     @Timed(value = "listener.participant_deregister_ack", description = "PARTICIPANT_DEREGISTER_ACK messages received")
 184     public void handleParticipantDeregisterAck(ParticipantDeregisterAck participantDeregisterAckMsg) {
 185         LOGGER.debug("ParticipantDeregisterAck message received as responseTo {}",
 186                 participantDeregisterAckMsg.getResponseTo());
 187     }
 188
 189     /**
 190      * Handle a ParticipantPrime message.
 191      *
 192      * @param participantPrimeMsg the ParticipantPrime message
 193      */
 194     @Timed(value = "listener.participant_prime", description = "PARTICIPANT_PRIME messages received")
 195     public void handleParticipantPrime(ParticipantPrime participantPrimeMsg) {
 196         LOGGER.debug("ParticipantPrime message received for participantId {}", participantPrimeMsg.getParticipantId());
 197         acDefinitionHandler.handlePrime(participantPrimeMsg);
 198     }
 199
 200     /**
 201      * Handle a ParticipantRestart message.
 202      *
 203      * @param participantRestartMsg the participantRestart message
 204      */
 205     @Timed(value = "listener.participant_restart", description = "PARTICIPANT_RESTART messages received")
 206     public void handleParticipantRestart(ParticipantRestart participantRestartMsg) {
 207         LOGGER.debug("ParticipantRestart message received for participantId {}",
 208                 participantRestartMsg.getParticipantId());
 209         acDefinitionHandler.handleParticipantRestart(participantRestartMsg);
 210     }
 211
 212     /**
 213      * Handle a ParticipantSync message.
 214      *
 215      * @param participantSyncMsg the participantSync message
 216      */
 217     @Timed(value = "listener.participant_sync_msg", description = "PARTICIPANT_SYNC messages received")
 218     public void handleParticipantSync(ParticipantSync participantSyncMsg) {
 219         LOGGER.debug("ParticipantSync message received for participantId {}",
 220                 participantSyncMsg.getParticipantId());
 221     }
 222
 223     /**
 224      * Dispatch a heartbeat for this participant.
 225      */
 226     public void sendHeartbeat() {
 227         if (publisher.isActive()) {
 228             if (!cacheProvider.isRegistered()) {
 229                 sendParticipantRegister();
 230             } else {
 231                 publisher.sendParticipantStatus(makeHeartbeat());
 232             }
 233         }
 234     }
 235
 236     /**
 237      * Method to send heartbeat to automation composition runtime.
 238      */
 239     private ParticipantStatus makeHeartbeat() {
 240         var heartbeat = new ParticipantStatus();
 241         heartbeat.setParticipantId(cacheProvider.getParticipantId());
 242         heartbeat.setReplicaId(cacheProvider.getReplicaId());
 243         heartbeat.setState(ParticipantState.ON_LINE);
 244         heartbeat.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 245
 246         return heartbeat;
 247     }
 248 }