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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2021-2025 OpenInfra Foundation Europe. All rights reserved.
   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.acm.participant.intermediary.handler.cache.AutomationCompositionMsg;
  29 import org.onap.policy.clamp.acm.participant.intermediary.handler.cache.CacheProvider;
  30 import org.onap.policy.clamp.models.acm.concepts.ParticipantState;
  31 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionDeploy;
  32 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionMigration;
  33 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionPrepare;
  34 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionStateChange;
  35 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantAckMessage;
  36 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregister;
  37 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantDeregisterAck;
  38 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantMessage;
  39 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrime;
  40 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegister;
  41 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRegisterAck;
  42 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatus;
  43 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatusReq;
  44 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
  45 import org.onap.policy.clamp.models.acm.messages.kafka.participant.PropertiesUpdate;
  46 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.DeployOrder;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49 import org.springframework.stereotype.Component;
  50
  51 /**
  52  * This class is responsible for managing the state of a participant.
  53  */
  54 @Component
  55 @RequiredArgsConstructor
  56 public class ParticipantHandler {
  57     private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantHandler.class);
  58
  59     private final AutomationCompositionHandler automationCompositionHandler;
  60     private final AcLockHandler acLockHandler;
  61     private final AcSubStateHandler acSubStateHandler;
  62     private final AcDefinitionHandler acDefinitionHandler;
  63     private final ParticipantMessagePublisher publisher;
  64     private final CacheProvider cacheProvider;
  65     private final MsgExecutor msgExecutor;
  66
  67     /**
  68      * Method which handles a participant health check event from clamp.
  69      *
  70      * @param participantStatusReqMsg participant participantStatusReq message
  71      */
  72     @Timed(value = "listener.participant_status_req", description = "PARTICIPANT_STATUS_REQ messages received")
  73     public void handleParticipantStatusReq(final ParticipantStatusReq participantStatusReqMsg) {
  74         sendHeartbeat();
  75     }
  76
  77     /**
  78      * Handle a automation composition update message.
  79      *
  80      * @param updateMsg the update message
  81      */
  82     @Timed(
  83             value = "listener.automation_composition_update",
  84             description = "AUTOMATION_COMPOSITION_UPDATE messages received")
  85     public void handleAutomationCompositionDeploy(AutomationCompositionDeploy updateMsg) {
  86         var acMsg = new AutomationCompositionMsg<>(
  87                 automationCompositionHandler::handleAutomationCompositionDeploy, updateMsg);
  88         setCompositionUpdate(updateMsg, acMsg);
  89         msgExecutor.execute(acMsg);
  90     }
  91
  92     /**
  93      * Handle a automation composition state change message.
  94      *
  95      * @param stateChangeMsg the state change message
  96      */
  97     @Timed(
  98             value = "listener.automation_composition_state_change",
  99             description = "AUTOMATION_COMPOSITION_STATE_CHANGE messages received")
 100     public void handleAutomationCompositionStateChange(AutomationCompositionStateChange stateChangeMsg) {
 101         var acMsg = DeployOrder.NONE.equals(stateChangeMsg.getDeployOrderedState())
 102                 ? new AutomationCompositionMsg<>(
 103                 acLockHandler::handleAutomationCompositionStateChange, stateChangeMsg)
 104                 : new AutomationCompositionMsg<>(
 105                 automationCompositionHandler::handleAutomationCompositionStateChange, stateChangeMsg);
 106         setCompositionUpdate(stateChangeMsg, acMsg);
 107         if (!DeployOrder.DELETE.equals(stateChangeMsg.getDeployOrderedState())) {
 108             setInstanceUpdate(stateChangeMsg, acMsg);
 109         }
 110         msgExecutor.execute(acMsg);
 111     }
 112
 113     /**
 114      * Handle a automation composition migration message.
 115      *
 116      * @param migrationMsg the migration message
 117      */
 118     @Timed(
 119             value = "listener.automation_composition_migration",
 120             description = "AUTOMATION_COMPOSITION_MIGRATION messages received")
 121     public void handleAutomationCompositionMigration(AutomationCompositionMigration migrationMsg) {
 122         var acMsg = Boolean.TRUE.equals(migrationMsg.getPrecheck())
 123                 ? new AutomationCompositionMsg<>(acSubStateHandler::handleAcMigrationPrecheck, migrationMsg)
 124                 : new AutomationCompositionMsg<>(
 125                         automationCompositionHandler::handleAutomationCompositionMigration, migrationMsg);
 126         setCompositionUpdate(migrationMsg, acMsg);
 127         setInstanceUpdate(migrationMsg, acMsg);
 128         acMsg.setCompositionTargetId(migrationMsg.getCompositionTargetId());
 129         acMsg.setRevisionIdCompositionTarget(migrationMsg.getRevisionIdCompositionTarget());
 130         msgExecutor.execute(acMsg);
 131     }
 132
 133     /**
 134      * Handle a automation composition property update message.
 135      *
 136      * @param propertyUpdateMsg the property update message
 137      */
 138     @Timed(value = "listener.properties_update", description = "PROPERTIES_UPDATE message received")
 139     public void handleAcPropertyUpdate(PropertiesUpdate propertyUpdateMsg) {
 140         var acMsg = new AutomationCompositionMsg<>(
 141                 automationCompositionHandler::handleAcPropertyUpdate, propertyUpdateMsg);
 142         setCompositionUpdate(propertyUpdateMsg, acMsg);
 143         setInstanceUpdate(propertyUpdateMsg, acMsg);
 144         msgExecutor.execute(acMsg);
 145     }
 146
 147     /**
 148      * Handle a automation composition Prepare/Review message.
 149      *
 150      * @param acPrepareMsg the AutomationComposition Prepare/Review message
 151      */
 152     @Timed(value = "listener.prepare", description = "AUTOMATION_COMPOSITION_PREPARE message received")
 153     public void handleAutomationCompositionPrepare(AutomationCompositionPrepare acPrepareMsg) {
 154         var acMsg = new AutomationCompositionMsg<>(
 155                 acSubStateHandler::handleAcPrepare, acPrepareMsg);
 156         setCompositionUpdate(acPrepareMsg, acMsg);
 157         if (!acPrepareMsg.isPreDeploy()) {
 158             setInstanceUpdate(acPrepareMsg, acMsg);
 159         }
 160         msgExecutor.execute(acMsg);
 161     }
 162
 163     private void setCompositionUpdate(ParticipantMessage participantMsg, AutomationCompositionMsg<?> acMsg) {
 164         acMsg.setCompositionId(participantMsg.getCompositionId());
 165         acMsg.setRevisionIdComposition(participantMsg.getRevisionIdComposition());
 166     }
 167
 168     private void setInstanceUpdate(ParticipantMessage participantMsg, AutomationCompositionMsg<?> acMsg) {
 169         acMsg.setInstanceId(participantMsg.getAutomationCompositionId());
 170         acMsg.setRevisionIdInstance(participantMsg.getRevisionIdInstance());
 171     }
 172
 173     /**
 174      * Check if a participant message applies to this participant handler.
 175      *
 176      * @param participantMsg the message to check
 177      * @return true if it applies, false otherwise
 178      */
 179     public boolean appliesTo(ParticipantMessage participantMsg) {
 180         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 181     }
 182
 183     /**
 184      * Check if a participant message applies to this participant handler.
 185      *
 186      * @param participantMsg the message to check
 187      * @return true if it applies, false otherwise
 188      */
 189     public boolean appliesTo(ParticipantAckMessage participantMsg) {
 190         return participantMsg.appliesTo(cacheProvider.getParticipantId(), cacheProvider.getReplicaId());
 191     }
 192
 193     /**
 194      * Method to send ParticipantRegister message to automation composition runtime.
 195      */
 196     public void sendParticipantRegister() {
 197         var participantRegister = new ParticipantRegister();
 198         participantRegister.setParticipantId(cacheProvider.getParticipantId());
 199         participantRegister.setReplicaId(cacheProvider.getReplicaId());
 200         participantRegister.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 201
 202         publisher.sendParticipantRegister(participantRegister);
 203     }
 204
 205     /**
 206      * Handle a participantRegister Ack message.
 207      *
 208      * @param participantRegisterAckMsg the participantRegisterAck message
 209      */
 210     @Timed(value = "listener.participant_register_ack", description = "PARTICIPANT_REGISTER_ACK messages received")
 211     public void handleParticipantRegisterAck(ParticipantRegisterAck participantRegisterAckMsg) {
 212         LOGGER.debug("ParticipantRegisterAck message received as responseTo {}",
 213                 participantRegisterAckMsg.getResponseTo());
 214         cacheProvider.setRegistered(true);
 215         publisher.sendParticipantStatus(makeHeartbeat());
 216     }
 217
 218     /**
 219      * Method to send ParticipantDeregister message to automation composition runtime.
 220      */
 221     public void sendParticipantDeregister() {
 222         var participantDeregister = new ParticipantDeregister();
 223         participantDeregister.setParticipantId(cacheProvider.getParticipantId());
 224         participantDeregister.setReplicaId(cacheProvider.getReplicaId());
 225         publisher.sendParticipantDeregister(participantDeregister);
 226     }
 227
 228     /**
 229      * Handle a participantDeregister Ack message.
 230      *
 231      * @param participantDeregisterAckMsg the participantDeregisterAck message
 232      */
 233     @Timed(value = "listener.participant_deregister_ack", description = "PARTICIPANT_DEREGISTER_ACK messages received")
 234     public void handleParticipantDeregisterAck(ParticipantDeregisterAck participantDeregisterAckMsg) {
 235         LOGGER.debug("ParticipantDeregisterAck message received as responseTo {}",
 236                 participantDeregisterAckMsg.getResponseTo());
 237     }
 238
 239     /**
 240      * Handle a ParticipantPrime message.
 241      *
 242      * @param participantPrimeMsg the ParticipantPrime message
 243      */
 244     @Timed(value = "listener.participant_prime", description = "PARTICIPANT_PRIME messages received")
 245     public void handleParticipantPrime(ParticipantPrime participantPrimeMsg) {
 246         LOGGER.debug("ParticipantPrime message received for participantId {}", participantPrimeMsg.getParticipantId());
 247         acDefinitionHandler.handlePrime(participantPrimeMsg);
 248     }
 249
 250     /**
 251      * Handle a ParticipantSync message.
 252      *
 253      * @param participantSyncMsg the participantSync message
 254      */
 255     @Timed(value = "listener.participant_sync_msg", description = "PARTICIPANT_SYNC messages received")
 256     public void handleParticipantSync(ParticipantSync participantSyncMsg) {
 257         if (participantSyncMsg.getExcludeReplicas().contains(cacheProvider.getReplicaId())) {
 258             LOGGER.debug("Ignore ParticipantSync message {}", participantSyncMsg.getMessageId());
 259             return;
 260         }
 261         LOGGER.debug("ParticipantSync message received for participantId {}", participantSyncMsg.getParticipantId());
 262         acDefinitionHandler.handleParticipantSync(participantSyncMsg);
 263         msgExecutor.check();
 264     }
 265
 266     /**
 267      * Dispatch a heartbeat for this participant.
 268      */
 269     public void sendHeartbeat() {
 270         if (publisher.isActive()) {
 271             if (!cacheProvider.isRegistered()) {
 272                 sendParticipantRegister();
 273             } else {
 274                 publisher.sendParticipantStatus(makeHeartbeat());
 275             }
 276         }
 277     }
 278
 279     /**
 280      * Method to send heartbeat to automation composition runtime.
 281      */
 282     private ParticipantStatus makeHeartbeat() {
 283         var heartbeat = new ParticipantStatus();
 284         heartbeat.setParticipantId(cacheProvider.getParticipantId());
 285         heartbeat.setReplicaId(cacheProvider.getReplicaId());
 286         heartbeat.setState(ParticipantState.ON_LINE);
 287         heartbeat.setParticipantSupportedElementType(cacheProvider.getSupportedAcElementTypes());
 288
 289         return heartbeat;
 290     }
 291 }