main/src/main/java/org/onap/policy/pap/main/comm/PdpStatusMessageHandler.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019-2021,2023 Nordix Foundation.
   4  *  Modifications Copyright (C) 2019-2021 AT&T Intellectual Property.
   5  *  Modifications Copyright (C) 2021-2022 Bell Canada. 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.pap.main.comm;
  24
  25 import java.sql.SQLIntegrityConstraintViolationException;
  26 import java.time.Instant;
  27 import java.util.Arrays;
  28 import java.util.Collections;
  29 import java.util.HashMap;
  30 import java.util.LinkedList;
  31 import java.util.List;
  32 import java.util.Map;
  33 import java.util.Optional;
  34 import java.util.concurrent.TimeUnit;
  35 import java.util.stream.Collectors;
  36 import org.apache.commons.lang3.builder.EqualsBuilder;
  37 import org.onap.policy.models.base.PfModelException;
  38 import org.onap.policy.models.pdp.concepts.Pdp;
  39 import org.onap.policy.models.pdp.concepts.PdpGroup;
  40 import org.onap.policy.models.pdp.concepts.PdpGroupFilter;
  41 import org.onap.policy.models.pdp.concepts.PdpStatistics;
  42 import org.onap.policy.models.pdp.concepts.PdpStatus;
  43 import org.onap.policy.models.pdp.concepts.PdpSubGroup;
  44 import org.onap.policy.models.pdp.enums.PdpState;
  45 import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
  46 import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
  47 import org.onap.policy.pap.main.PolicyPapException;
  48 import org.onap.policy.pap.main.parameters.PapParameterGroup;
  49 import org.onap.policy.pap.main.parameters.PdpParameters;
  50 import org.onap.policy.pap.main.service.PdpGroupService;
  51 import org.onap.policy.pap.main.service.PdpStatisticsService;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54 import org.springframework.stereotype.Component;
  55
  56
  57 /**
  58  * Handler for PDP Status messages which either represent registration or heart beat.
  59  *
  60  * @author Ram Krishna Verma (ram.krishna.verma@est.tech)
  61  */
  62
  63 @Component
  64 public class PdpStatusMessageHandler extends PdpMessageGenerator {
  65     private static final Logger LOGGER = LoggerFactory.getLogger(PdpStatusMessageHandler.class);
  66
  67     private final PdpParameters params;
  68
  69     private final boolean savePdpStatistics;
  70
  71     private final PdpGroupService pdpGroupService;
  72
  73     private final PdpStatisticsService pdpStatisticsService;
  74
  75     /**
  76      * List to store policies present in db.
  77      */
  78     List<ToscaPolicy> policies = new LinkedList<>();
  79
  80     /**
  81      * List to store policies to be deployed (heartbeat).
  82      */
  83     Map<ToscaConceptIdentifier, ToscaPolicy> policiesToBeDeployed = new HashMap<>();
  84
  85     /**
  86      * List to store policies to be undeployed (heartbeat).
  87      */
  88     List<ToscaConceptIdentifier> policiesToBeUndeployed = new LinkedList<>();
  89
  90     /**
  91      * Constructs the object.
  92      *
  93      * @param parameterGroup the parameterGroup
  94      * @param pdpGroupService the pdpGroupService
  95      * @param pdpStatisticsService the pdpStatisticsService
  96      */
  97     public PdpStatusMessageHandler(PapParameterGroup parameterGroup, PdpGroupService pdpGroupService,
  98         PdpStatisticsService pdpStatisticsService) {
  99         super(true);
 100         this.params = parameterGroup.getPdpParameters();
 101         this.savePdpStatistics = parameterGroup.isSavePdpStatisticsInDb();
 102         this.pdpGroupService = pdpGroupService;
 103         this.pdpStatisticsService = pdpStatisticsService;
 104     }
 105
 106     /**
 107      * Handles the PdpStatus message coming from various PDP's.
 108      *
 109      * @param message the PdpStatus message
 110      */
 111     public void handlePdpStatus(final PdpStatus message) {
 112         if (message.getPolicies() == null) {
 113             message.setPolicies(Collections.emptyList());
 114         }
 115
 116         long diffms = System.currentTimeMillis() - message.getTimestampMs();
 117         if (diffms > params.getMaxMessageAgeMs()) {
 118             long diffsec = TimeUnit.SECONDS.convert(diffms, TimeUnit.MILLISECONDS);
 119             LOGGER.info("discarding status message from {} age {}s", message.getName(), diffsec);
 120             return;
 121         }
 122
 123         synchronized (updateLock) {
 124             try {
 125                 if (message.getPdpSubgroup() == null) {
 126                     handlePdpRegistration(message);
 127                 } else {
 128                     handlePdpHeartbeat(message);
 129                 }
 130             } catch (final PolicyPapException exp) {
 131                 LOGGER.error("Operation Failed", exp);
 132             } catch (final Exception exp) {
 133                 if (isDuplicateKeyException(exp, Exception.class)) {
 134                     /*
 135                      * this is to be expected, if multiple PAPs are processing the same
 136                      * heartbeat at a time, thus we log the exception at a trace level
 137                      * instead of an error level.
 138                      */
 139                     LOGGER.info("Failed updating PDP information for {} - may have been added by another PAP",
 140                                     message.getName());
 141                     LOGGER.trace("Failed updating PDP information for {}", message.getName(), exp);
 142                 } else {
 143                     LOGGER.error("Failed connecting to database provider", exp);
 144                 }
 145             }
 146         }
 147     }
 148
 149     /**
 150      * Determines if the exception indicates a duplicate key.
 151      *
 152      * @param thrown exception to check
 153      * @param exceptionClazz the class to check against
 154      * @return {@code true} if the exception occurred due to a duplicate key
 155      */
 156     protected static boolean isDuplicateKeyException(Throwable thrown, Class<? extends Throwable> exceptionClazz) {
 157         while (thrown != null) {
 158             if (thrown instanceof SQLIntegrityConstraintViolationException) {
 159                 return true;
 160             }
 161
 162             if (exceptionClazz.isInstance(thrown)) {
 163                 if (isDuplicateKeyException(thrown.getCause(), exceptionClazz)) {
 164                     return true;
 165                 }
 166             }
 167
 168             thrown = thrown.getCause();
 169         }
 170
 171         return false;
 172     }
 173
 174     private void handlePdpRegistration(final PdpStatus message) throws PfModelException, PolicyPapException {
 175         if (!findAndUpdatePdpGroup(message)) {
 176             final var errorMessage = "Failed to register PDP. No matching PdpGroup/SubGroup Found - ";
 177             LOGGER.debug("{}{}", errorMessage, message);
 178             throw new PolicyPapException(errorMessage + message);
 179         }
 180     }
 181
 182     private boolean findAndUpdatePdpGroup(final PdpStatus message)
 183             throws PfModelException {
 184         var pdpGroupFound = false;
 185         final PdpGroupFilter filter =
 186                 PdpGroupFilter.builder().name(message.getPdpGroup()).groupState(PdpState.ACTIVE).build();
 187
 188         final List<PdpGroup> pdpGroups = pdpGroupService.getFilteredPdpGroups(filter);
 189         if (!pdpGroups.isEmpty()) {
 190             pdpGroupFound = registerPdp(message, pdpGroups.get(0));
 191         }
 192         return pdpGroupFound;
 193     }
 194
 195     private boolean registerPdp(final PdpStatus message, final PdpGroup finalizedPdpGroup) throws PfModelException {
 196         Optional<PdpSubGroup> subGroup;
 197         var pdpGroupFound = false;
 198         subGroup = findPdpSubGroup(message, finalizedPdpGroup);
 199
 200         if (subGroup.isPresent()) {
 201             policies = getToscaPolicies(subGroup.get());
 202             policiesToBeDeployed = policies.stream().collect(Collectors
 203                     .toMap(ToscaPolicy::getIdentifier, policy -> policy));
 204             policiesToBeUndeployed = null;
 205
 206             LOGGER.debug("Found pdpGroup - {}, going for registration of PDP - {}", finalizedPdpGroup, message);
 207             Optional<Pdp> pdp = findPdpInstance(message, subGroup.get());
 208             if (pdp.isPresent()) {
 209                 updatePdpHealthStatus(message, subGroup.get(), pdp.get(), finalizedPdpGroup);
 210             } else {
 211                 updatePdpSubGroup(finalizedPdpGroup, subGroup.get(), message);
 212             }
 213             sendPdpMessage(finalizedPdpGroup.getName(), subGroup.get(), message.getName(), null);
 214             pdpGroupFound = true;
 215         }
 216         return pdpGroupFound;
 217     }
 218
 219     private void updatePdpSubGroup(final PdpGroup pdpGroup, final PdpSubGroup pdpSubGroup, final PdpStatus message) {
 220
 221         final var pdpInstance = new Pdp();
 222         pdpInstance.setInstanceId(message.getName());
 223         pdpInstance.setPdpState(PdpState.ACTIVE);
 224         pdpInstance.setHealthy(message.getHealthy());
 225         pdpInstance.setMessage(message.getDescription());
 226         pdpInstance.setLastUpdate(Instant.now());
 227         pdpSubGroup.getPdpInstances().add(pdpInstance);
 228
 229         pdpSubGroup.setCurrentInstanceCount(pdpSubGroup.getCurrentInstanceCount() + 1);
 230
 231         pdpGroupService.updatePdpSubGroup(pdpGroup.getName(), pdpSubGroup);
 232
 233         LOGGER.debug("Updated PdpSubGroup in DB - {} belonging to PdpGroup - {}", pdpSubGroup, pdpGroup.getName());
 234     }
 235
 236     private void handlePdpHeartbeat(final PdpStatus message) throws PfModelException {
 237
 238         final PdpGroupFilter filter =
 239                 PdpGroupFilter.builder().name(message.getPdpGroup()).groupState(PdpState.ACTIVE).build();
 240         final List<PdpGroup> pdpGroups = pdpGroupService.getFilteredPdpGroups(filter);
 241         if (!pdpGroups.isEmpty()) {
 242             var pdpGroup = pdpGroups.get(0);
 243             Optional<PdpSubGroup> pdpSubgroup = findPdpSubGroup(message, pdpGroup);
 244             if (pdpSubgroup.isPresent()) {
 245                 Optional<Pdp> pdpInstance = findPdpInstance(message, pdpSubgroup.get());
 246                 if (pdpInstance.isPresent()) {
 247                     processPdpDetails(message, pdpSubgroup.get(), pdpInstance.get(), pdpGroup);
 248                 } else {
 249                     LOGGER.debug("PdpInstance not Found in DB. Sending Pdp for registration - {}", message);
 250                     registerPdp(message, pdpGroup);
 251                 }
 252             }
 253         }
 254     }
 255
 256     private Optional<PdpSubGroup> findPdpSubGroup(final PdpStatus message, final PdpGroup pdpGroup) {
 257         PdpSubGroup pdpSubgroup = null;
 258         for (final PdpSubGroup subGroup : pdpGroup.getPdpSubgroups()) {
 259             if (message.getPdpType().equals(subGroup.getPdpType())) {
 260                 pdpSubgroup = subGroup;
 261                 break;
 262             }
 263         }
 264         return Optional.ofNullable(pdpSubgroup);
 265     }
 266
 267     private Optional<Pdp> findPdpInstance(final PdpStatus message, final PdpSubGroup subGroup) {
 268         Pdp pdpInstance = null;
 269         for (final Pdp pdpInstanceDetails : subGroup.getPdpInstances()) {
 270             if (pdpInstanceDetails.getInstanceId().equals(message.getName())) {
 271                 pdpInstance = pdpInstanceDetails;
 272                 break;
 273             }
 274         }
 275         return Optional.ofNullable(pdpInstance);
 276     }
 277
 278     private void processPdpDetails(final PdpStatus message, final PdpSubGroup pdpSubGroup, final Pdp pdpInstance,
 279             final PdpGroup pdpGroup) throws PfModelException {
 280         // all policies
 281         policies = getToscaPolicies(pdpSubGroup);
 282
 283         Map<ToscaConceptIdentifier, ToscaPolicy> policyMap =
 284                         policies.stream().collect(Collectors.toMap(ToscaPolicy::getIdentifier, policy -> policy));
 285
 286         // policies that the PDP already has (-) all
 287         policiesToBeUndeployed = message.getPolicies().stream().filter(policyId -> !policyMap.containsKey(policyId))
 288                         .collect(Collectors.toList());
 289
 290         // all (-) policies that the PDP already has
 291         policiesToBeDeployed = policyMap;
 292         policiesToBeDeployed.keySet().removeAll(message.getPolicies());
 293
 294         if (PdpState.TERMINATED.equals(message.getState())) {
 295             processPdpTermination(pdpSubGroup, pdpInstance, pdpGroup);
 296         } else if (validatePdpDetails(message, pdpGroup, pdpSubGroup, pdpInstance)) {
 297             LOGGER.debug("PdpInstance details are correct. Saving current state in DB - {}", pdpInstance);
 298             updatePdpHealthStatus(message, pdpSubGroup, pdpInstance, pdpGroup);
 299
 300             if (savePdpStatistics) {
 301                 processPdpStatistics(message, pdpSubGroup, pdpInstance, pdpGroup);
 302             } else {
 303                 LOGGER.debug("Not processing PdpStatistics - {}", message.getStatistics());
 304             }
 305         } else {
 306             LOGGER.debug("PdpInstance details are not correct. Sending PdpUpdate message - {}", pdpInstance);
 307             LOGGER.debug("Policy list in DB - {}. Policy list in heartbeat - {}", pdpSubGroup.getPolicies(),
 308                 message.getPolicies());
 309             updatePdpHealthStatus(message, pdpSubGroup, pdpInstance, pdpGroup);
 310             sendPdpMessage(pdpGroup.getName(), pdpSubGroup, pdpInstance.getInstanceId(), pdpInstance.getPdpState());
 311         }
 312     }
 313
 314     private void processPdpStatistics(final PdpStatus message, final PdpSubGroup pdpSubGroup, final Pdp pdpInstance,
 315                     final PdpGroup pdpGroup) {
 316         if (validatePdpStatisticsDetails(message, pdpInstance, pdpGroup, pdpSubGroup)) {
 317             LOGGER.debug("PdpStatistics details are correct. Saving current statistics in DB - {}",
 318                     message.getStatistics());
 319             createPdpStatistics(message.getStatistics());
 320         } else {
 321             LOGGER.debug("PdpStatistics details are not correct - {}", message.getStatistics());
 322         }
 323     }
 324
 325     private void processPdpTermination(final PdpSubGroup pdpSubGroup, final Pdp pdpInstance, final PdpGroup pdpGroup) {
 326         pdpSubGroup.getPdpInstances().remove(pdpInstance);
 327         pdpSubGroup.setCurrentInstanceCount(pdpSubGroup.getCurrentInstanceCount() - 1);
 328         pdpGroupService.updatePdpSubGroup(pdpGroup.getName(), pdpSubGroup);
 329
 330         LOGGER.debug("Deleted PdpInstance - {} belonging to PdpSubGroup - {} and PdpGroup - {}", pdpInstance,
 331                 pdpSubGroup, pdpGroup);
 332     }
 333
 334     private boolean validatePdpDetails(final PdpStatus message, final PdpGroup pdpGroup, final PdpSubGroup subGroup,
 335             final Pdp pdpInstanceDetails) {
 336         /*
 337          * "EqualsBuilder" is a bit of a misnomer, as it uses containsAll() to check policies. Nevertheless, it does the
 338          * job and provides a convenient way to build a bunch of comparisons.
 339          */
 340         return new EqualsBuilder().append(message.getPdpGroup(), pdpGroup.getName())
 341                 .append(message.getPdpSubgroup(), subGroup.getPdpType())
 342                 .append(message.getPdpType(), subGroup.getPdpType())
 343                 .append(message.getState(), pdpInstanceDetails.getPdpState())
 344                 .append(message.getPolicies().containsAll(subGroup.getPolicies()), true)
 345                 .append(subGroup.getPolicies().containsAll(message.getPolicies()), true).build();
 346     }
 347
 348     private boolean validatePdpStatisticsDetails(final PdpStatus message, final Pdp pdpInstanceDetails,
 349             final PdpGroup pdpGroup, final PdpSubGroup pdpSubGroup) {
 350         if (message.getStatistics() != null) {
 351             return new EqualsBuilder()
 352                     .append(message.getStatistics().getPdpInstanceId(), pdpInstanceDetails.getInstanceId())
 353                     .append(message.getStatistics().getPdpGroupName(), pdpGroup.getName())
 354                     .append(message.getStatistics().getPdpSubGroupName(), pdpSubGroup.getPdpType())
 355                     .append(message.getStatistics().getPolicyDeployCount() < 0, false)
 356                     .append(message.getStatistics().getPolicyDeployFailCount() < 0, false)
 357                     .append(message.getStatistics().getPolicyDeploySuccessCount() < 0, false)
 358                     .append(message.getStatistics().getPolicyUndeployCount() < 0, false)
 359                     .append(message.getStatistics().getPolicyUndeployFailCount() < 0, false)
 360                     .append(message.getStatistics().getPolicyUndeploySuccessCount() < 0, false)
 361                     .append(message.getStatistics().getPolicyExecutedCount() < 0, false)
 362                     .append(message.getStatistics().getPolicyExecutedFailCount() < 0, false)
 363                     .append(message.getStatistics().getPolicyExecutedSuccessCount() < 0, false).build();
 364         } else {
 365             LOGGER.debug("PdpStatistics is null");
 366             return false;
 367         }
 368     }
 369
 370     private void updatePdpHealthStatus(final PdpStatus message, final PdpSubGroup pdpSubgroup, final Pdp pdpInstance,
 371             final PdpGroup pdpGroup) {
 372         pdpInstance.setHealthy(message.getHealthy());
 373         pdpInstance.setMessage(message.getDescription());
 374         pdpInstance.setLastUpdate(Instant.now());
 375         pdpGroupService.updatePdp(pdpGroup.getName(), pdpSubgroup.getPdpType(), pdpInstance);
 376
 377         LOGGER.debug("Updated Pdp in DB - {}", pdpInstance);
 378     }
 379
 380     private void createPdpStatistics(final PdpStatistics pdpStatistics) {
 381         pdpStatisticsService.createPdpStatistics(Arrays.asList(pdpStatistics));
 382         LOGGER.debug("Created PdpStatistics in DB - {}", pdpStatistics);
 383     }
 384
 385     private void sendPdpMessage(final String pdpGroupName, final PdpSubGroup subGroup, final String pdpInstanceId,
 386             final PdpState pdpState) {
 387         final List<ToscaPolicy> polsToBeDeployed = new LinkedList<>(policiesToBeDeployed.values());
 388         final var pdpUpdatemessage =
 389             createPdpUpdateMessage(pdpGroupName, subGroup, pdpInstanceId,
 390                         polsToBeDeployed, policiesToBeUndeployed);
 391         final var pdpStateChangeMessage =
 392             createPdpStateChangeMessage(pdpGroupName, subGroup, pdpInstanceId, pdpState);
 393         updateDeploymentStatus(pdpGroupName, subGroup.getPdpType(), pdpInstanceId, pdpStateChangeMessage.getState(),
 394             pdpUpdatemessage.getPoliciesToBeDeployed());
 395
 396         requestMap.addRequest(pdpUpdatemessage, pdpStateChangeMessage);
 397         LOGGER.debug("Sent PdpUpdate message - {}", pdpUpdatemessage);
 398         LOGGER.debug("Sent PdpStateChange message - {}", pdpStateChangeMessage);
 399     }
 400 }