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

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