* ============LICENSE_START=======================================================
* ONAP PAP
* ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2021 Nordix Foundation.
+ * Modifications Copyright (C) 2021 Bell Canada. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.policy.pap.main.comm;
+import java.time.Instant;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.pap.concepts.PolicyNotification;
import org.onap.policy.models.pdp.concepts.Pdp;
import org.onap.policy.models.pdp.concepts.PdpGroup;
import org.onap.policy.models.pdp.concepts.PdpGroupFilter;
import org.onap.policy.models.pdp.concepts.PdpMessage;
import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
import org.onap.policy.models.pdp.concepts.PdpSubGroup;
import org.onap.policy.models.pdp.concepts.PdpUpdate;
import org.onap.policy.models.pdp.enums.PdpState;
import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
import org.onap.policy.pap.main.comm.msgdata.Request;
import org.onap.policy.pap.main.comm.msgdata.RequestListener;
import org.onap.policy.pap.main.comm.msgdata.StateChangeReq;
import org.onap.policy.pap.main.comm.msgdata.UpdateReq;
+import org.onap.policy.pap.main.notification.DeploymentStatus;
import org.onap.policy.pap.main.notification.PolicyNotifier;
import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
import org.onap.policy.pap.main.parameters.RequestParams;
*/
private final PolicyNotifier policyNotifier;
+ /**
+ * Used to undeploy policies from the system, when they cannot be deployed to a PDP.
+ *
+ * <p/>
+ * Note: there's a "catch-22" here. The request map needs an undeployer, but the
+ * undeployer needs the request map. Consequently, the request map is created first,
+ * then the undeployer, and finally, this field is set.
+ */
+ @Setter
+ private PolicyUndeployer policyUndeployer;
+
/**
* Constructs the object.
* Adds an UPDATE request to the map.
*
* @param update the UPDATE request or {@code null}
+ * @return the new request (this should only be used by junit tests)
*/
- public void addRequest(PdpUpdate update) {
+ public Request addRequest(PdpUpdate update) {
if (update == null) {
- return;
+ return null;
}
if (isBroadcast(update)) {
// @formatter:on
String name = update.getName() + " " + PdpUpdate.class.getSimpleName();
- UpdateReq request = new UpdateReq(reqparams, name, update);
+ var request = new UpdateReq(reqparams, name, update);
addSingleton(request);
+ return request;
}
/**
* Adds a STATE-CHANGE request to the map.
*
* @param stateChange the STATE-CHANGE request or {@code null}
+ * @return the new request (this should only be used by junit tests)
*/
- public void addRequest(PdpStateChange stateChange) {
+ public Request addRequest(PdpStateChange stateChange) {
if (stateChange == null) {
- return;
+ return null;
}
if (isBroadcast(stateChange)) {
// @formatter:on
String name = stateChange.getName() + " " + PdpStateChange.class.getSimpleName();
- StateChangeReq request = new StateChangeReq(reqparams, name, stateChange);
+ var request = new StateChangeReq(reqparams, name, stateChange);
addSingleton(request);
+ return request;
}
/**
}
/**
- * Removes a PDP from all active groups.
- *
- * @param pdpName name of the PDP to be removed
- * @return {@code true} if the PDP was removed from a group, {@code false} if it was
- * not assigned to a group
- * @throws PfModelException if an error occurred
+ * Removes expired PDPs from all active groups.
*/
- public boolean removeFromGroups(String pdpName) throws PfModelException {
+ public void removeExpiredPdps() {
+
+ synchronized (modifyLock) {
+ logger.info("check for PDP records older than {}ms", params.getMaxPdpAgeMs());
+
+ try (PolicyModelsProvider dao = daoFactory.create()) {
- try (PolicyModelsProvider dao = daoFactory.create()) {
+ PdpGroupFilter filter = PdpGroupFilter.builder().groupState(PdpState.ACTIVE).build();
+ List<PdpGroup> groups = dao.getFilteredPdpGroups(filter);
+ List<PdpGroup> updates = new ArrayList<>(1);
- PdpGroupFilter filter = PdpGroupFilter.builder().groupState(PdpState.ACTIVE).build();
- List<PdpGroup> groups = dao.getFilteredPdpGroups(filter);
- List<PdpGroup> updates = new ArrayList<>(1);
+ var status = new DeploymentStatus(dao);
- for (PdpGroup group : groups) {
- if (removeFromGroup(pdpName, group)) {
- updates.add(group);
+ Instant minAge = Instant.now().minusMillis(params.getMaxPdpAgeMs());
+
+ for (PdpGroup group : groups) {
+ Set<String> pdps = removeFromGroup(minAge, group);
+ if (!pdps.isEmpty()) {
+ updates.add(group);
+ status.loadByGroup(group.getName());
+ pdps.forEach(status::deleteDeployment);
+ }
}
- }
- if (updates.isEmpty()) {
- return false;
+ if (!updates.isEmpty()) {
+ dao.updatePdpGroups(updates);
- } else {
- dao.updatePdpGroups(updates);
- return true;
+ var notification = new PolicyNotification();
+ status.flush(notification);
+
+ policyNotifier.publish(notification);
+ }
+
+ } catch (PfModelException e) {
+ logger.warn("failed to remove expired PDPs", e);
}
}
}
/**
- * Removes a PDP from a group.
+ * Removes expired PDPs from a group.
*
- * @param pdpName name of the PDP to be removed
- * @param group group from which it should be removed
- * @return {@code true} if the PDP was removed from the group, {@code false} if it was
- * not assigned to the group
+ * @param minAge minimum age for active PDPs
+ * @param group group from which expired PDPs should be removed
+ * @return the expired PDPs
*/
- private boolean removeFromGroup(String pdpName, PdpGroup group) {
+ private Set<String> removeFromGroup(Instant minAge, PdpGroup group) {
+ Set<String> pdps = new HashSet<>();
for (PdpSubGroup subgrp : group.getPdpSubgroups()) {
- if (removeFromSubgroup(pdpName, group, subgrp)) {
- return true;
- }
+ removeFromSubgroup(minAge, group, subgrp, pdps);
}
- return false;
+ return pdps;
}
/**
- * Removes a PDP from a subgroup.
+ * Removes expired PDPs from a subgroup.
*
- * @param pdpName name of the PDP to be removed
+ * @param minAge minimum age for active PDPs
* @param group group from which to attempt to remove the PDP
* @param subgrp subgroup from which to attempt to remove the PDP
- * @return {@code true} if the PDP was removed, {@code false} if the PDP was not in
- * the group
+ * @param pdps where to place the expired PDPs
*/
- private boolean removeFromSubgroup(String pdpName, PdpGroup group, PdpSubGroup subgrp) {
+ private void removeFromSubgroup(Instant minAge, PdpGroup group, PdpSubGroup subgrp, Set<String> pdps) {
Iterator<Pdp> iter = subgrp.getPdpInstances().iterator();
while (iter.hasNext()) {
Pdp instance = iter.next();
- if (pdpName.equals(instance.getInstanceId())) {
+ if (instance.getLastUpdate().isBefore(minAge)) {
+ String pdpName = instance.getInstanceId();
logger.info("removed {} from group={} subgroup={}", pdpName, group.getName(), subgrp.getPdpType());
iter.remove();
subgrp.setCurrentInstanceCount(subgrp.getPdpInstances().size());
- return true;
+ pdps.add(pdpName);
}
}
-
- return false;
}
/**
return new PdpRequests(pdpName, policyNotifier);
}
+ /**
+ * Makes a handler for PDP responses.
+ * @return a response handler
+ */
+ protected PdpStatusMessageHandler makePdpResponseHandler() {
+ return new PdpStatusMessageHandler(params.getParams(), params.isSavePdpStatistics());
+ }
+
/**
* Listener for singleton request events.
*/
private class SingletonListener implements RequestListener {
private final PdpRequests requests;
private final Request request;
+ private final String pdpName;
public SingletonListener(PdpRequests requests, Request request) {
this.requests = requests;
this.request = request;
+ this.pdpName = requests.getPdpName();
}
@Override
- public void failure(String pdpName, String reason) {
- if (requests.getPdpName().equals(pdpName)) {
- disablePdp(requests);
+ public void failure(String responsePdpName, String reason) {
+ Collection<ToscaConceptIdentifier> undeployPolicies = requestCompleted(responsePdpName);
+ if (undeployPolicies.isEmpty()) {
+ // nothing to undeploy
+ return;
}
- }
- @Override
- public void success(String pdpName) {
- if (requests.getPdpName().equals(pdpName)) {
- if (pdp2requests.get(requests.getPdpName()) == requests) {
- startNextRequest(requests, request);
-
- } else {
- logger.info("discard old requests for {}", pdpName);
- requests.stopPublishing();
- }
+ /*
+ * Undeploy the extra policies. Note: this will likely cause a new message to
+ * be assigned to the request, thus we must re-start it after making the
+ * change.
+ */
+ PdpMessage oldmsg = request.getMessage();
+
+ try {
+ logger.warn("undeploy policies from {}:{} that failed to deploy: {}", oldmsg.getPdpGroup(),
+ oldmsg.getPdpSubgroup(), undeployPolicies);
+ policyUndeployer.undeploy(oldmsg.getPdpGroup(), oldmsg.getPdpSubgroup(), undeployPolicies);
+ } catch (PfModelException | RuntimeException e) {
+ logger.error("cannot undeploy policies {}", undeployPolicies, e);
+ }
+
+ if (request.getMessage() == oldmsg) {
+ // message is unchanged - start the next request
+ startNextRequest(request);
+ } else {
+ // message changed - restart the request
+ request.startPublishing();
}
}
@Override
- public void retryCountExhausted() {
- disablePdp(requests);
- }
+ public void success(String responsePdpName, PdpStatus response) {
+ requestCompleted(responsePdpName);
- /**
- * Starts the next request associated with a PDP.
- *
- * @param requests current set of requests
- * @param request the request that just completed
- */
- private void startNextRequest(PdpRequests requests, Request request) {
- if (!requests.startNextRequest(request)) {
- pdp2requests.remove(requests.getPdpName(), requests);
+ if (!(request instanceof UpdateReq)) {
+ // other response types may not include the list of policies
+ return;
}
+
+ /*
+ * Update PDP time stamps. Also send pdp-update and pdp-state-change, as
+ * necessary, if the response does not reflect what's in the DB.
+ */
+ var handler = makePdpResponseHandler();
+ handler.handlePdpStatus(response);
}
/**
- * Disables a PDP by removing it from its subgroup and then sending it a PASSIVE
- * request.
+ * Handles a request completion, starting the next request, if there is one.
*
- * @param requests the requests associated with the PDP to be disabled
+ * @param responsePdpName name of the PDP provided in the response
+ * @return a list of policies to be undeployed
*/
- private void disablePdp(PdpRequests requests) {
-
- policyNotifier.removePdp(requests.getPdpName());
-
- // remove the requests from the map
- if (!pdp2requests.remove(requests.getPdpName(), requests)) {
- // don't have the info we need to disable it
- logger.warn("no requests with which to disable {}", requests.getPdpName());
- return;
+ private Collection<ToscaConceptIdentifier> requestCompleted(String responsePdpName) {
+ if (!pdpName.equals(responsePdpName)) {
+ return Collections.emptyList();
}
- logger.warn("disabling {}", requests.getPdpName());
-
- requests.stopPublishing();
+ if (pdp2requests.get(pdpName) != requests) {
+ logger.info("discard old requests for {}", responsePdpName);
+ requests.stopPublishing();
+ return Collections.emptyList();
+ }
- // remove the PDP from all groups
- boolean removed = false;
- try {
- removed = removeFromGroups(requests.getPdpName());
- } catch (PfModelException e) {
- logger.info("unable to remove PDP {} from subgroup", requests.getPdpName(), e);
+ if (!requests.isFirstInQueue(request)) {
+ logger.error("request is not first in the queue {}", request.getMessage());
+ return Collections.emptyList();
}
- // send the state change
- PdpStateChange change = new PdpStateChange();
- change.setName(requests.getPdpName());
- change.setState(PdpState.PASSIVE);
+ Collection<ToscaConceptIdentifier> undeployPolicies = request.getUndeployPolicies();
+ if (undeployPolicies.isEmpty()) {
+ // nothing to undeploy - just start the next request
+ startNextRequest(request);
+ }
- if (removed) {
- // send an update, too
- PdpUpdate update = new PdpUpdate();
- update.setName(requests.getPdpName());
- update.setPolicies(Collections.emptyList());
+ return undeployPolicies;
+ }
- addRequest(update, change);
+ @Override
+ public void retryCountExhausted(Request request) {
+ if (pdp2requests.get(pdpName) == requests) {
+ requests.stopPublishing();
+ startNextRequest(request);
+ }
+ }
- } else {
- addRequest(change);
+ /**
+ * Starts the next request associated with a PDP.
+ *
+ * @param request the request that just completed
+ */
+ private void startNextRequest(Request request) {
+ if (!requests.startNextRequest(request)) {
+ pdp2requests.remove(pdpName, requests);
}
}
}
Code Review / policy / pap.git / blobdiff