import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.onap.policy.controlloop.actorserviceprovider.CallbackManager;
import org.onap.policy.controlloop.actorserviceprovider.Operation;
import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.OperationProperties;
+import org.onap.policy.controlloop.actorserviceprovider.OperationResult;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
import org.onap.policy.controlloop.actorserviceprovider.parameters.OperatorConfig;
import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
-import org.onap.policy.controlloop.policy.PolicyResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
* Partial implementation of an operator. In general, it's preferable that subclasses
* would override {@link #startOperationAsync(int, OperationOutcome)
* startOperationAsync()}. However, if that proves to be too difficult, then they can
- * simply override {@link #doOperation(int, OperationOutcome) doOperation()}. In addition,
- * if the operation requires any preprocessor steps, the subclass may choose to override
- * {@link #startPreprocessorAsync()}.
+ * simply override {@link #doOperation(int, OperationOutcome) doOperation()}.
* <p/>
* The futures returned by the methods within this class can be canceled, and will
* propagate the cancellation to any subtasks. Thus it is also expected that any futures
return config.getBlockingExecutor();
}
+ @Override
public String getActorName() {
return params.getActor();
}
+ @Override
public String getName() {
return params.getOperation();
}
- /**
- * Determines if a property has been assigned for the operation.
- *
- * @param name property name
- * @return {@code true} if the given property has been assigned for the operation,
- * {@code false} otherwise
- */
+ @Override
public boolean containsProperty(String name) {
return properties.containsKey(name);
}
- /**
- * Sets a property.
- *
- * @param name property name
- * @param value new value
- */
+ @Override
public void setProperty(String name, Object value) {
+ logger.info("{}: set property {}={}", getFullName(), name, value);
properties.put(name, value);
}
- /**
- * Gets a property's value.
- *
- * @param name name of the property of interest
- * @return the property's value, or {@code null} if it has no value
- */
@SuppressWarnings("unchecked")
+ @Override
public <T> T getProperty(String name) {
return (T) properties.get(name);
}
- @Override
- public CompletableFuture<OperationOutcome> start() {
- // allocate a controller for the entire operation
- final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
-
- CompletableFuture<OperationOutcome> preproc = startPreprocessorAsync();
- if (preproc == null) {
- // no preprocessor required - just start the operation
- return startOperationAttempt(controller, 1);
- }
-
- /*
- * Do preprocessor first and then, if successful, start the operation. Note:
- * operations create their own outcome, ignoring the outcome from any previous
- * steps.
- *
- * Wrap the preprocessor to ensure "stop" is propagated to it.
- */
- // @formatter:off
- controller.wrap(preproc)
- .exceptionally(fromException("preprocessor of operation"))
- .thenCompose(handlePreprocessorFailure(controller))
- .thenCompose(unusedOutcome -> startOperationAttempt(controller, 1))
- .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
- // @formatter:on
-
- return controller;
- }
-
- /**
- * Handles a failure in the preprocessor pipeline. If a failure occurred, then it
- * invokes the call-backs, marks the controller complete, and returns an incomplete
- * future, effectively halting the pipeline. Otherwise, it returns the outcome that it
- * received.
- * <p/>
- * Assumes that no callbacks have been invoked yet.
- *
- * @param controller pipeline controller
- * @return a function that checks the outcome status and continues, if successful, or
- * indicates a failure otherwise
- */
- private Function<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
- PipelineControllerFuture<OperationOutcome> controller) {
-
- return outcome -> {
-
- if (isSuccess(outcome)) {
- logger.info("{}: preprocessor succeeded for {}", getFullName(), params.getRequestId());
- return CompletableFuture.completedFuture(outcome);
- }
-
- logger.warn("preprocessor failed, discontinuing operation {} for {}", getFullName(), params.getRequestId());
-
- final Executor executor = params.getExecutor();
- final CallbackManager callbacks = new CallbackManager();
-
- // propagate "stop" to the callbacks
- controller.add(callbacks);
-
- final OperationOutcome outcome2 = params.makeOutcome();
-
- // TODO need a FAILURE_MISSING_DATA (e.g., A&AI)
-
- outcome2.setFinalOutcome(true);
- outcome2.setResult(PolicyResult.FAILURE_GUARD);
- outcome2.setMessage(outcome != null ? outcome.getMessage() : null);
-
- // @formatter:off
- CompletableFuture.completedFuture(outcome2)
- .whenCompleteAsync(callbackStarted(callbacks), executor)
- .whenCompleteAsync(callbackCompleted(callbacks), executor)
- .whenCompleteAsync(controller.delayedComplete(), executor);
- // @formatter:on
-
- return new CompletableFuture<>();
- };
- }
-
/**
- * Invokes the operation's preprocessor step(s) as a "future". This method simply
- * returns {@code null}.
- * <p/>
- * This method assumes the following:
- * <ul>
- * <li>the operator is alive</li>
- * <li>exceptions generated within the pipeline will be handled by the invoker</li>
- * </ul>
+ * Gets a property value, throwing an exception if it's missing.
*
- * @return a function that will start the preprocessor and returns its outcome, or
- * {@code null} if this operation needs no preprocessor
- */
- protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
- return null;
- }
-
- /**
- * Invokes the operation's guard step(s) as a "future".
- * <p/>
- * This method assumes the following:
- * <ul>
- * <li>the operator is alive</li>
- * <li>exceptions generated within the pipeline will be handled by the invoker</li>
- * </ul>
- *
- * @return a function that will start the guard checks and returns its outcome, or
- * {@code null} if this operation has no guard
+ * @param name property name
+ * @param propertyType property type, used in an error message if the property value
+ * is {@code null}
+ * @return the property value
*/
- protected CompletableFuture<OperationOutcome> startGuardAsync() {
- if (params.isPreprocessed()) {
- return null;
+ @SuppressWarnings("unchecked")
+ protected <T> T getRequiredProperty(String name, String propertyType) {
+ T value = (T) properties.get(name);
+ if (value == null) {
+ throw new IllegalStateException("missing " + propertyType);
}
- // get the guard payload
- Map<String, Object> payload = makeGuardPayload();
-
- /*
- * Note: can't use constants from actor.guard, because that would create a
- * circular dependency.
- */
- return params.toBuilder().actor(GUARD_ACTOR_NAME).operation(GUARD_OPERATION_NAME).retry(null).timeoutSec(null)
- .payload(payload).build().start();
+ return value;
}
- /**
- * Creates a payload to execute a guard operation.
- *
- * @return a new guard payload
- */
- protected Map<String, Object> makeGuardPayload() {
- // TODO delete this once preprocessing is done by the application
- Map<String, Object> guard = new LinkedHashMap<>();
- guard.put("actor", params.getActor());
- guard.put("operation", params.getOperation());
- guard.put("target", params.getTargetEntity());
- guard.put("requestId", params.getRequestId());
-
- String clname = params.getContext().getEvent().getClosedLoopControlName();
- if (clname != null) {
- guard.put("clname", clname);
- }
+ @Override
+ public CompletableFuture<OperationOutcome> start() {
+ // allocate a controller for the entire operation
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
- return guard;
+ // start attempt #1
+ return startOperationAttempt(controller, 1);
}
/**
- * Starts the operation attempt, with no preprocessor. When all retries complete, it
- * will complete the controller.
+ * Starts the operation attempt. When all retries complete, it will complete the
+ * controller.
*
* @param controller controller for all operation attempts
* @param attempt attempt number, typically starting with 1
logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
final Executor executor = params.getExecutor();
- final OperationOutcome outcome = params.makeOutcome();
+ final OperationOutcome outcome = makeOutcome();
final CallbackManager callbacks = new CallbackManager();
// this operation attempt gets its own controller
* @return {@code true} if the outcome was successful
*/
protected boolean isSuccess(OperationOutcome outcome) {
- return (outcome != null && outcome.getResult() == PolicyResult.SUCCESS);
+ return (outcome != null && outcome.getResult() == OperationResult.SUCCESS);
}
/**
* <i>and</i> was associated with this operator, {@code false} otherwise
*/
protected boolean isActorFailed(OperationOutcome outcome) {
- return (isSameOperation(outcome) && outcome.getResult() == PolicyResult.FAILURE);
+ return (isSameOperation(outcome) && outcome.getResult() == OperationResult.FAILURE);
}
/**
outcome = origOutcome;
} else {
logger.warn("{}: null outcome; treating as a failure for {}", getFullName(), params.getRequestId());
- outcome = this.setOutcome(params.makeOutcome(), PolicyResult.FAILURE);
+ outcome = this.setOutcome(makeOutcome(), OperationResult.FAILURE);
}
// ensure correct actor/operation
outcome.setOperation(getName());
// determine if we should retry, based on the result
- if (outcome.getResult() != PolicyResult.FAILURE) {
+ if (outcome.getResult() != OperationResult.FAILURE) {
// do not retry success or other failure types (e.g., exception)
outcome.setFinalOutcome(true);
return outcome;
* FAILURE_RETRIES
*/
logger.info("operation {} retries exhausted for {}", getFullName(), params.getRequestId());
- outcome.setResult(PolicyResult.FAILURE_RETRIES);
+ outcome.setResult(OperationResult.FAILURE_RETRIES);
outcome.setFinalOutcome(true);
}
private Function<Throwable, OperationOutcome> fromException(String type) {
return thrown -> {
- OperationOutcome outcome = params.makeOutcome();
+ OperationOutcome outcome = makeOutcome();
if (thrown instanceof CancellationException || thrown.getCause() instanceof CancellationException) {
// do not include exception in the message, as it just clutters the log
* @return the updated operation
*/
public OperationOutcome setOutcome(OperationOutcome operation, Throwable thrown) {
- PolicyResult result = (isTimeout(thrown) ? PolicyResult.FAILURE_TIMEOUT : PolicyResult.FAILURE_EXCEPTION);
+ OperationResult result = (isTimeout(thrown) ? OperationResult.FAILURE_TIMEOUT
+ : OperationResult.FAILURE_EXCEPTION);
return setOutcome(operation, result);
}
* @param result result of the operation
* @return the updated operation
*/
- public OperationOutcome setOutcome(OperationOutcome operation, PolicyResult result) {
+ public OperationOutcome setOutcome(OperationOutcome operation, OperationResult result) {
logger.trace("{}: set outcome {} for {}", getFullName(), result, params.getRequestId());
operation.setResult(result);
- operation.setMessage(result == PolicyResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
+ operation.setMessage(result == OperationResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
: ControlLoopOperation.FAILED_MSG);
return operation;
}
+ /**
+ * Makes an outcome, populating the "target" field with the contents of the target
+ * entity property.
+ *
+ * @return a new operation outcome
+ */
+ protected OperationOutcome makeOutcome() {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setTarget(getProperty(OperationProperties.AAI_TARGET_ENTITY));
+ return outcome;
+ }
+
/**
* Determines if a throwable is due to a timeout.
*
Code Review / policy / models.git / blobdiff