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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP PAP
   4  * ================================================================================
   5  * Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
   6  * Modifications Copyright (C) 2021 Nordix Foundation.
   7  * Modifications Copyright (C) 2021-2022 Bell Canada. All rights reserved.
   8  * ================================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.policy.pap.main.comm;
  24
  25 import java.time.Instant;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.Collections;
  29 import java.util.HashMap;
  30 import java.util.HashSet;
  31 import java.util.Iterator;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Set;
  35 import org.onap.policy.models.base.PfModelException;
  36 import org.onap.policy.models.pap.concepts.PolicyNotification;
  37 import org.onap.policy.models.pdp.concepts.Pdp;
  38 import org.onap.policy.models.pdp.concepts.PdpGroup;
  39 import org.onap.policy.models.pdp.concepts.PdpGroupFilter;
  40 import org.onap.policy.models.pdp.concepts.PdpMessage;
  41 import org.onap.policy.models.pdp.concepts.PdpStateChange;
  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.concepts.PdpUpdate;
  45 import org.onap.policy.models.pdp.enums.PdpState;
  46 import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
  47 import org.onap.policy.pap.main.comm.msgdata.Request;
  48 import org.onap.policy.pap.main.comm.msgdata.RequestListener;
  49 import org.onap.policy.pap.main.comm.msgdata.StateChangeReq;
  50 import org.onap.policy.pap.main.comm.msgdata.UpdateReq;
  51 import org.onap.policy.pap.main.notification.DeploymentStatus;
  52 import org.onap.policy.pap.main.notification.PolicyNotifier;
  53 import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
  54 import org.onap.policy.pap.main.parameters.RequestParams;
  55 import org.onap.policy.pap.main.service.PdpGroupService;
  56 import org.onap.policy.pap.main.service.PolicyStatusService;
  57 import org.slf4j.Logger;
  58 import org.slf4j.LoggerFactory;
  59 import org.springframework.stereotype.Component;
  60
  61 /**
  62  * Maps a PDP name to requests that modify PDPs.
  63  */
  64 @Component
  65 public class PdpModifyRequestMap {
  66     private static final Logger logger = LoggerFactory.getLogger(PdpModifyRequestMap.class);
  67
  68     private static final String UNEXPECTED_BROADCAST = "unexpected broadcast message: ";
  69
  70     /**
  71      * Maps a PDP name to its outstanding requests.
  72      */
  73     private final Map<String, PdpRequests> pdp2requests = new HashMap<>();
  74
  75     /**
  76      * PDP modification lock.
  77      */
  78     private Object modifyLock;
  79
  80     /**
  81      * The configuration parameters.
  82      */
  83     private PdpModifyRequestMapParams params;
  84
  85     /**
  86      * Used to notify when policy updates completes.
  87      */
  88     private final PolicyNotifier policyNotifier;
  89
  90     /**
  91      * Used to undeploy policies from the system, when they cannot be deployed to a PDP.
  92      *
  93      * <p/>
  94      * Note: The request map needs an undeployer during creation, and the undeployer
  95      * needs the request map when it's initialize method is called.
  96      */
  97     private final PolicyUndeployer policyUndeployer;
  98
  99     private final PdpGroupService pdpGroupService;
 100
 101     private final PolicyStatusService policyStatusService;
 102
 103     private final PdpStatusMessageHandler pdpStatusMessageHandler;
 104
 105     /**
 106      * Constructs the object.
 107      *
 108      * @param pdpGroupService the pdpGroupService
 109      * @param policyStatusService the policyStatusService
 110      * @param pdpStatusMessageHandler the pdpStatusMessageHandler
 111      * @param policyUndeployer the policyUndeployer
 112      * @param policyNotifier the policyNotifier
 113      */
 114     public PdpModifyRequestMap(PdpGroupService pdpGroupService, PolicyStatusService policyStatusService,
 115         PdpStatusMessageHandler pdpStatusMessageHandler, PolicyUndeployer policyUndeployer,
 116         PolicyNotifier policyNotifier) {
 117         this.pdpGroupService = pdpGroupService;
 118         this.policyStatusService = policyStatusService;
 119         this.pdpStatusMessageHandler = pdpStatusMessageHandler;
 120         this.policyUndeployer = policyUndeployer;
 121         this.policyNotifier = policyNotifier;
 122     }
 123
 124     /**
 125      * Initializes the requestMap.
 126      *
 127      * @param params the parameters.
 128      */
 129     public void initialize(PdpModifyRequestMapParams params) {
 130         params.validate();
 131
 132         this.params = params;
 133         this.modifyLock = params.getModifyLock();
 134     }
 135
 136     /**
 137      * Determines if the map contains any requests.
 138      *
 139      * @return {@code true} if the map is empty, {@code false} otherwise
 140      */
 141     public boolean isEmpty() {
 142         return pdp2requests.isEmpty();
 143     }
 144
 145     /**
 146      * Stops publishing requests to the given PDP.
 147      *
 148      * @param pdpName PDP name
 149      */
 150     public void stopPublishing(String pdpName) {
 151         synchronized (modifyLock) {
 152             PdpRequests requests = pdp2requests.remove(pdpName);
 153             if (requests != null) {
 154                 requests.stopPublishing();
 155             }
 156         }
 157     }
 158
 159     /**
 160      * Adds a pair of requests to the map.
 161      *
 162      * @param update the UPDATE request or {@code null}
 163      * @param stateChange the STATE-CHANGE request or {@code null}
 164      */
 165     public void addRequest(PdpUpdate update, PdpStateChange stateChange) {
 166         if (update == null) {
 167             addRequest(stateChange);
 168
 169         } else if (stateChange == null) {
 170             addRequest(update);
 171
 172         } else if (stateChange.getState() == PdpState.ACTIVE) {
 173             // publish update before activating
 174             synchronized (modifyLock) {
 175                 addRequest(update);
 176                 addRequest(stateChange);
 177             }
 178
 179         } else {
 180             // deactivate before publishing update
 181             synchronized (modifyLock) {
 182                 addRequest(stateChange);
 183                 addRequest(update);
 184             }
 185         }
 186     }
 187
 188     /**
 189      * Adds an UPDATE request to the map.
 190      *
 191      * @param update the UPDATE request or {@code null}
 192      * @return the new request (this should only be used by junit tests)
 193      */
 194     public Request addRequest(PdpUpdate update) {
 195         if (update == null) {
 196             return null;
 197         }
 198
 199         if (isBroadcast(update)) {
 200             throw new IllegalArgumentException(UNEXPECTED_BROADCAST + update);
 201         }
 202
 203         // @formatter:off
 204         RequestParams reqparams = new RequestParams()
 205             .setMaxRetryCount(params.getParams().getUpdateParameters().getMaxRetryCount())
 206             .setTimers(params.getUpdateTimers())
 207             .setModifyLock(params.getModifyLock())
 208             .setPdpPublisher(params.getPdpPublisher())
 209             .setResponseDispatcher(params.getResponseDispatcher());
 210         // @formatter:on
 211
 212         String name = update.getName() + " " + PdpUpdate.class.getSimpleName();
 213         var request = new UpdateReq(reqparams, name, update);
 214
 215         addSingleton(request);
 216         return request;
 217     }
 218
 219     /**
 220      * Adds a STATE-CHANGE request to the map.
 221      *
 222      * @param stateChange the STATE-CHANGE request or {@code null}
 223      * @return the new request (this should only be used by junit tests)
 224      */
 225     public Request addRequest(PdpStateChange stateChange) {
 226         if (stateChange == null) {
 227             return null;
 228         }
 229
 230         if (isBroadcast(stateChange)) {
 231             throw new IllegalArgumentException(UNEXPECTED_BROADCAST + stateChange);
 232         }
 233
 234         // @formatter:off
 235         RequestParams reqparams = new RequestParams()
 236             .setMaxRetryCount(params.getParams().getStateChangeParameters().getMaxRetryCount())
 237             .setTimers(params.getStateChangeTimers())
 238             .setModifyLock(params.getModifyLock())
 239             .setPdpPublisher(params.getPdpPublisher())
 240             .setResponseDispatcher(params.getResponseDispatcher());
 241         // @formatter:on
 242
 243         String name = stateChange.getName() + " " + PdpStateChange.class.getSimpleName();
 244         var request = new StateChangeReq(reqparams, name, stateChange);
 245
 246         addSingleton(request);
 247         return request;
 248     }
 249
 250     /**
 251      * Determines if a message is a broadcast message.
 252      *
 253      * @param message the message to examine
 254      * @return {@code true} if the message is a broadcast message, {@code false} if
 255      *         destined for a single PDP
 256      */
 257     private boolean isBroadcast(PdpMessage message) {
 258         return (message.getName() == null);
 259     }
 260
 261     /**
 262      * Configures and adds a request to the map.
 263      *
 264      * @param request the request to be added
 265      */
 266     private void addSingleton(Request request) {
 267
 268         synchronized (modifyLock) {
 269             PdpRequests requests = pdp2requests.computeIfAbsent(request.getMessage().getName(), this::makePdpRequests);
 270
 271             request.setListener(new SingletonListener(requests, request));
 272             requests.addSingleton(request);
 273         }
 274     }
 275
 276     /**
 277      * Removes expired PDPs from all active groups.
 278      */
 279     public void removeExpiredPdps() {
 280
 281         synchronized (modifyLock) {
 282             logger.info("check for PDP records older than {}ms", params.getMaxPdpAgeMs());
 283
 284             try {
 285
 286                 PdpGroupFilter filter = PdpGroupFilter.builder().groupState(PdpState.ACTIVE).build();
 287                 List<PdpGroup> groups = pdpGroupService.getFilteredPdpGroups(filter);
 288                 List<PdpGroup> updates = new ArrayList<>(1);
 289
 290                 var status = new DeploymentStatus(policyStatusService);
 291
 292                 Instant minAge = Instant.now().minusMillis(params.getMaxPdpAgeMs());
 293
 294                 for (PdpGroup group : groups) {
 295                     Set<String> pdps = removeFromGroup(minAge, group);
 296                     if (!pdps.isEmpty()) {
 297                         updates.add(group);
 298                         status.loadByGroup(group.getName());
 299                         pdps.forEach(status::deleteDeployment);
 300                     }
 301                 }
 302
 303                 if (!updates.isEmpty()) {
 304                     pdpGroupService.updatePdpGroups(updates);
 305
 306                     var notification = new PolicyNotification();
 307                     status.flush(notification);
 308
 309                     policyNotifier.publish(notification);
 310                 }
 311
 312             } catch (RuntimeException e) {
 313                 logger.warn("failed to remove expired PDPs", e);
 314             }
 315         }
 316     }
 317
 318     /**
 319      * Removes expired PDPs from a group.
 320      *
 321      * @param minAge minimum age for active PDPs
 322      * @param group group from which expired PDPs should be removed
 323      * @return the expired PDPs
 324      */
 325     private Set<String> removeFromGroup(Instant minAge, PdpGroup group) {
 326         Set<String> pdps = new HashSet<>();
 327         for (PdpSubGroup subgrp : group.getPdpSubgroups()) {
 328             removeFromSubgroup(minAge, group, subgrp, pdps);
 329         }
 330
 331         return pdps;
 332     }
 333
 334     /**
 335      * Removes expired PDPs from a subgroup.
 336      *
 337      * @param minAge minimum age for active PDPs
 338      * @param group group from which to attempt to remove the PDP
 339      * @param subgrp subgroup from which to attempt to remove the PDP
 340      * @param pdps where to place the expired PDPs
 341      */
 342     private void removeFromSubgroup(Instant minAge, PdpGroup group, PdpSubGroup subgrp, Set<String> pdps) {
 343
 344         Iterator<Pdp> iter = subgrp.getPdpInstances().iterator();
 345
 346         while (iter.hasNext()) {
 347             Pdp instance = iter.next();
 348
 349             if (instance.getLastUpdate().isBefore(minAge)) {
 350                 String pdpName = instance.getInstanceId();
 351                 logger.info("removed {} from group={} subgroup={}", pdpName, group.getName(), subgrp.getPdpType());
 352                 iter.remove();
 353                 subgrp.setCurrentInstanceCount(subgrp.getPdpInstances().size());
 354                 pdps.add(pdpName);
 355             }
 356         }
 357     }
 358
 359     /**
 360      * Creates a new set of requests for a PDP. May be overridden by junit tests.
 361      *
 362      * @param pdpName PDP name
 363      * @return a new set of requests
 364      */
 365     protected PdpRequests makePdpRequests(String pdpName) {
 366         return new PdpRequests(pdpName, policyNotifier);
 367     }
 368
 369     /**
 370      * Listener for singleton request events.
 371      */
 372     private class SingletonListener implements RequestListener {
 373         private final PdpRequests requests;
 374         private final Request request;
 375         private final String pdpName;
 376
 377         public SingletonListener(PdpRequests requests, Request request) {
 378             this.requests = requests;
 379             this.request = request;
 380             this.pdpName = requests.getPdpName();
 381         }
 382
 383         @Override
 384         public void failure(String responsePdpName, String reason) {
 385             Collection<ToscaConceptIdentifier> undeployPolicies = requestCompleted(responsePdpName);
 386             if (undeployPolicies.isEmpty()) {
 387                 // nothing to undeploy
 388                 return;
 389             }
 390
 391             /*
 392              * Undeploy the extra policies. Note: this will likely cause a new message to
 393              * be assigned to the request, thus we must re-start it after making the
 394              * change.
 395              */
 396             PdpMessage oldmsg = request.getMessage();
 397
 398             try {
 399                 logger.warn("undeploy policies from {}:{} that failed to deploy: {}", oldmsg.getPdpGroup(),
 400                                 oldmsg.getPdpSubgroup(), undeployPolicies);
 401                 policyUndeployer.undeploy(oldmsg.getPdpGroup(), oldmsg.getPdpSubgroup(), undeployPolicies);
 402             } catch (PfModelException | RuntimeException e) {
 403                 logger.error("cannot undeploy policies {}", undeployPolicies, e);
 404             }
 405
 406             if (request.getMessage() == oldmsg) {
 407                 // message is unchanged - start the next request
 408                 startNextRequest(request);
 409             } else {
 410                 // message changed - restart the request
 411                 request.startPublishing();
 412             }
 413         }
 414
 415         @Override
 416         public void success(String responsePdpName, PdpStatus response) {
 417             requestCompleted(responsePdpName);
 418
 419             if (!(request instanceof UpdateReq)) {
 420                 // other response types may not include the list of policies
 421                 return;
 422             }
 423
 424             /*
 425              * Update PDP time stamps. Also send pdp-update and pdp-state-change, as
 426              * necessary, if the response does not reflect what's in the DB.
 427              */
 428             pdpStatusMessageHandler.handlePdpStatus(response);
 429         }
 430
 431         /**
 432          * Handles a request completion, starting the next request, if there is one.
 433          *
 434          * @param responsePdpName name of the PDP provided in the response
 435          * @return a list of policies to be undeployed
 436          */
 437         private Collection<ToscaConceptIdentifier> requestCompleted(String responsePdpName) {
 438             if (!pdpName.equals(responsePdpName)) {
 439                 return Collections.emptyList();
 440             }
 441
 442             if (pdp2requests.get(pdpName) != requests) {
 443                 logger.info("discard old requests for {}", responsePdpName);
 444                 requests.stopPublishing();
 445                 return Collections.emptyList();
 446             }
 447
 448             if (!requests.isFirstInQueue(request)) {
 449                 logger.error("request is not first in the queue {}", request.getMessage());
 450                 return Collections.emptyList();
 451             }
 452
 453             Collection<ToscaConceptIdentifier> undeployPolicies = request.getUndeployPolicies();
 454             if (undeployPolicies.isEmpty()) {
 455                 // nothing to undeploy - just start the next request
 456                 startNextRequest(request);
 457             }
 458
 459             return undeployPolicies;
 460         }
 461
 462         @Override
 463         public void retryCountExhausted(Request request) {
 464             if (pdp2requests.get(pdpName) == requests) {
 465                 requests.stopPublishing();
 466                 startNextRequest(request);
 467             }
 468         }
 469
 470         /**
 471          * Starts the next request associated with a PDP.
 472          *
 473          * @param request the request that just completed
 474          */
 475         private void startNextRequest(Request request) {
 476             if (!requests.startNextRequest(request)) {
 477                 pdp2requests.remove(pdpName, requests);
 478             }
 479         }
 480     }
 481 }