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