* ============LICENSE_START=======================================================
* ONAP
* ================================================================================
- * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2020-2022 AT&T Intellectual Property. 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.
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.Setter;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
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
* returned by overridden methods will do the same. Of course, if a class overrides
* {@link #doOperation(int, OperationOutcome) doOperation()}, then there's little that can
* be done to cancel that particular operation.
+ * <p/>
+ * In general tasks in a pipeline are executed by the same thread. However, the following
+ * should always be executed via the executor specified in "params":
+ * <ul>
+ * <li>start callback</li>
+ * <li>completion callback</li>
+ * <li>controller completion (i.e., delayedComplete())</li>
+ * </ul>
*/
public abstract class OperationPartial implements Operation {
private static final Logger logger = LoggerFactory.getLogger(OperationPartial.class);
private static final Coder coder = new StandardCoder();
+ public static final String GUARD_ACTOR_NAME = "GUARD";
+ public static final String GUARD_OPERATION_NAME = "Decision";
public static final long DEFAULT_RETRY_WAIT_MS = 1000L;
- // values extracted from the operator
-
- private final OperatorPartial operator;
+ private final OperatorConfig config;
/**
* Operation parameters.
*/
protected final ControlLoopOperationParams params;
+ @Getter
+ private final String fullName;
+
+ @Getter
+ @Setter(AccessLevel.PROTECTED)
+ private String subRequestId;
+
+ @Getter
+ private final List<String> propertyNames;
+
+ /**
+ * Values for the properties identified by {@link #getPropertyNames()}.
+ */
+ private final Map<String, Object> properties = new HashMap<>();
+
/**
* Constructs the object.
*
* @param params operation parameters
- * @param operator operator that created this operation
+ * @param config configuration for this operation
+ * @param propertyNames names of properties required by this operation
*/
- public OperationPartial(ControlLoopOperationParams params, OperatorPartial operator) {
+ protected OperationPartial(ControlLoopOperationParams params, OperatorConfig config, List<String> propertyNames) {
this.params = params;
- this.operator = operator;
+ this.config = config;
+ this.fullName = params.getActor() + "." + params.getOperation();
+ this.propertyNames = propertyNames;
}
public Executor getBlockingExecutor() {
- return operator.getBlockingExecutor();
- }
-
- public String getFullName() {
- return operator.getFullName();
+ return config.getBlockingExecutor();
}
+ @Override
public String getActorName() {
- return operator.getActorName();
+ return params.getActor();
}
+ @Override
public String getName() {
- return operator.getName();
+ return params.getOperation();
}
@Override
- public final CompletableFuture<OperationOutcome> start() {
- if (!operator.isAlive()) {
- throw new IllegalStateException("operation is not running: " + getFullName());
- }
-
- // 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);
- }
+ public boolean containsProperty(String name) {
+ return properties.containsKey(name);
+ }
- /*
- * 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
+ @Override
+ public void setProperty(String name, Object value) {
+ logger.info("{}: set property {}={}", getFullName(), name, value);
+ properties.put(name, value);
+ }
- return controller;
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T getProperty(String name) {
+ return (T) properties.get(name);
}
/**
- * 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.
+ * Gets a property value, throwing an exception if it's missing.
*
- * @param controller pipeline controller
- * @return a function that checks the outcome status and continues, if successful, or
- * indicates a failure otherwise
+ * @param name property name
+ * @param propertyType property type, used in an error message if the property value
+ * is {@code null}
+ * @return the property value
*/
- private Function<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
- PipelineControllerFuture<OperationOutcome> controller) {
-
- return outcome -> {
-
- if (outcome != null && 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.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
+ @SuppressWarnings("unchecked")
+ public <T> T getRequiredProperty(String name, String propertyType) {
+ T value = (T) properties.get(name);
+ if (value == null) {
+ throw new IllegalStateException("missing " + propertyType);
+ }
- return new CompletableFuture<>();
- };
+ return value;
}
- /**
- * Invokes the operation's preprocessor step(s) as a "future". This method simply
- * invokes {@link #startGuardAsync()}.
- * <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 preprocessor and returns its outcome, or
- * {@code null} if this operation needs no preprocessor
- */
- protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
- return startGuardAsync();
- }
+ @Override
+ public CompletableFuture<OperationOutcome> start() {
+ // allocate a controller for the entire operation
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
- /**
- * Invokes the operation's guard 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>
- *
- * @return a function that will start the guard checks and returns its outcome, or
- * {@code null} if this operation has no guard
- */
- protected CompletableFuture<OperationOutcome> startGuardAsync() {
- return null;
+ // 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
private CompletableFuture<OperationOutcome> startOperationAttempt(
PipelineControllerFuture<OperationOutcome> controller, int attempt) {
+ generateSubRequestId(attempt);
+
// propagate "stop" to the operation attempt
controller.wrap(startAttemptWithoutRetries(attempt)).thenCompose(retryOnFailure(controller, attempt))
.whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
return controller;
}
+ /**
+ * Generates and sets {@link #subRequestId} to a new subrequest ID.
+ *
+ * @param attempt attempt number, typically starting with 1
+ */
+ public void generateSubRequestId(int attempt) {
+ // Note: this should be "protected", but that makes junits much messier
+
+ setSubRequestId(UUID.randomUUID().toString());
+ }
+
/**
* Starts the operation attempt, without doing any retries.
*
- * @param params operation parameters
* @param attempt attempt number, typically starting with 1
* @return a future that will return the result of a single operation attempt
*/
logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
- final Executor executor = params.getExecutor();
- final OperationOutcome outcome = params.makeOutcome();
- final CallbackManager callbacks = new CallbackManager();
+ final var executor = params.getExecutor();
+ final var outcome = makeOutcome();
+ final var callbacks = new CallbackManager();
// this operation attempt gets its own controller
final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
* @return {@code true} if the outcome was successful
*/
protected boolean isSuccess(OperationOutcome outcome) {
- return (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);
}
/**
*/
private Function<OperationOutcome, OperationOutcome> setRetryFlag(int attempt) {
- return operation -> {
- if (operation != null && !isActorFailed(operation)) {
- /*
- * wrong type or wrong operation - just leave it as is. No need to log
- * anything here, as retryOnFailure() will log a message
- */
- return operation;
+ return origOutcome -> {
+ // ensure we have a non-null outcome
+ OperationOutcome outcome;
+ if (origOutcome != null) {
+ outcome = origOutcome;
+ } else {
+ logger.warn("{}: null outcome; treating as a failure for {}", getFullName(), params.getRequestId());
+ outcome = this.setOutcome(makeOutcome(), OperationResult.FAILURE);
}
- // get a non-null operation
- OperationOutcome oper2;
- if (operation != null) {
- oper2 = operation;
- } else {
- oper2 = params.makeOutcome();
- oper2.setResult(PolicyResult.FAILURE);
+ // ensure correct actor/operation
+ outcome.setActor(getActorName());
+ outcome.setOperation(getName());
+
+ // determine if we should retry, based on the result
+ if (outcome.getResult() != OperationResult.FAILURE) {
+ // do not retry success or other failure types (e.g., exception)
+ outcome.setFinalOutcome(true);
+ return outcome;
}
int retry = getRetry(params.getRetry());
- if (retry > 0 && attempt > retry) {
+ if (retry <= 0) {
+ // no retries were specified
+ outcome.setFinalOutcome(true);
+
+ } else if (attempt <= retry) {
+ // have more retries - not the final outcome
+ outcome.setFinalOutcome(false);
+
+ } else {
/*
- * retries were specified and we've already tried them all - change to
+ * retries were specified, and we've already tried them all - change to
* FAILURE_RETRIES
*/
logger.info("operation {} retries exhausted for {}", getFullName(), params.getRequestId());
- oper2.setResult(PolicyResult.FAILURE_RETRIES);
+ outcome.setResult(OperationResult.FAILURE_RETRIES);
+ outcome.setFinalOutcome(true);
}
- return oper2;
+ return outcome;
};
}
private Function<Throwable, OperationOutcome> fromException(String type) {
return thrown -> {
- OperationOutcome outcome = params.makeOutcome();
+ OperationOutcome outcome = makeOutcome();
- logger.warn("exception throw by {} {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
- params.getRequestId(), thrown);
+ if (thrown instanceof CancellationException || thrown.getCause() instanceof CancellationException) {
+ // do not include exception in the message, as it just clutters the log
+ logger.warn("{} canceled {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
+ params.getRequestId());
+ } else {
+ logger.warn("exception thrown by {} {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
+ params.getRequestId(), thrown);
+ }
return setOutcome(outcome, thrown);
};
* created. If this future is canceled, then all of the futures will be
* canceled
*/
- protected CompletableFuture<OperationOutcome> anyOf(
+ public CompletableFuture<OperationOutcome> anyOf(
@SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
return anyOf(Arrays.asList(futureMakers));
* canceled. Similarly, when this future completes, any incomplete futures
* will be canceled
*/
- protected CompletableFuture<OperationOutcome> anyOf(
- List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
+ public CompletableFuture<OperationOutcome> anyOf(List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
return futures[0];
}
- CompletableFuture.anyOf(futures).thenApply(outcome -> (OperationOutcome) outcome)
+ CompletableFuture.anyOf(futures).thenApply(OperationOutcome.class::cast)
.whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
return controller;
* created. If this future is canceled, then all of the futures will be
* canceled
*/
- protected CompletableFuture<OperationOutcome> allOf(
+ public CompletableFuture<OperationOutcome> allOf(
@SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
return allOf(Arrays.asList(futureMakers));
* canceled. Similarly, when this future completes, any incomplete futures
* will be canceled
*/
- protected CompletableFuture<OperationOutcome> allOf(
- List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
+ public CompletableFuture<OperationOutcome> allOf(List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
Queue<OperationOutcome> outcomes = new LinkedList<>();
* @return an array of futures, possibly zero-length. If the array is of size one,
* then that one item should be returned instead of the controller
*/
+ @SuppressWarnings("unchecked")
private CompletableFuture<OperationOutcome>[] attachFutures(PipelineControllerFuture<OperationOutcome> controller,
List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers,
UnaryOperator<CompletableFuture<OperationOutcome>> adorn) {
if (futureMakers.isEmpty()) {
- @SuppressWarnings("unchecked")
- CompletableFuture<OperationOutcome>[] result = new CompletableFuture[0];
- return result;
+ return new CompletableFuture[0];
}
// the last, unadorned future that is created
}
}
- @SuppressWarnings("unchecked")
- CompletableFuture<OperationOutcome>[] result = new CompletableFuture[futures.size()];
+ var result = new CompletableFuture[futures.size()];
if (result.length == 1) {
// special case - return the unadorned future
* @param futureMakers functions to make the futures
* @return a future to cancel the sequence or await the outcome
*/
- protected CompletableFuture<OperationOutcome> sequence(
+ public CompletableFuture<OperationOutcome> sequence(
@SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
return sequence(Arrays.asList(futureMakers));
* @return a future to cancel the sequence or await the outcome, or {@code null} if
* there were no tasks to perform
*/
- protected CompletableFuture<OperationOutcome> sequence(
+ public CompletableFuture<OperationOutcome> sequence(
List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
Queue<Supplier<CompletableFuture<OperationOutcome>>> queue = new ArrayDeque<>(futureMakers);
* executing
*/
final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
- final Executor executor = params.getExecutor();
+ final var executor = params.getExecutor();
// @formatter:off
controller.wrap(nextTask)
- .thenComposeAsync(nextTaskOnSuccess(controller, queue), executor)
+ .thenCompose(nextTaskOnSuccess(controller, queue))
.whenCompleteAsync(controller.delayedComplete(), executor);
// @formatter:on
// @formatter:off
return controller
.wrap(nextTask)
- .thenComposeAsync(nextTaskOnSuccess(controller, taskQueue), params.getExecutor());
+ .thenCompose(nextTaskOnSuccess(controller, taskQueue));
// @formatter:on
};
}
* @param callbacks used to determine if the start callback can be invoked
* @return a function that sets the start time and invokes the callback
*/
- private BiConsumer<OperationOutcome, Throwable> callbackStarted(CallbackManager callbacks) {
+ protected BiConsumer<OperationOutcome, Throwable> callbackStarted(CallbackManager callbacks) {
return (outcome, thrown) -> {
if (callbacks.canStart()) {
- // haven't invoked "start" callback yet
+ outcome.setSubRequestId(getSubRequestId());
outcome.setStart(callbacks.getStartTime());
outcome.setEnd(null);
- params.callbackStarted(outcome);
+
+ // pass a copy to the callback
+ var outcome2 = new OperationOutcome(outcome);
+ outcome2.setFinalOutcome(false);
+ params.callbackStarted(outcome2);
}
};
}
* @param callbacks used to determine if the end callback can be invoked
* @return a function that sets the end time and invokes the callback
*/
- private BiConsumer<OperationOutcome, Throwable> callbackCompleted(CallbackManager callbacks) {
+ protected BiConsumer<OperationOutcome, Throwable> callbackCompleted(CallbackManager callbacks) {
return (outcome, thrown) -> {
-
if (callbacks.canEnd()) {
+ outcome.setSubRequestId(getSubRequestId());
outcome.setStart(callbacks.getStartTime());
outcome.setEnd(callbacks.getEndTime());
- params.callbackCompleted(outcome);
+
+ // pass a copy to the callback
+ params.callbackCompleted(new OperationOutcome(outcome));
}
};
}
* @param operation operation to be updated
* @return the updated operation
*/
- protected OperationOutcome setOutcome(OperationOutcome operation, Throwable thrown) {
- PolicyResult result = (isTimeout(thrown) ? PolicyResult.FAILURE_TIMEOUT : PolicyResult.FAILURE_EXCEPTION);
+ public OperationOutcome setOutcome(OperationOutcome operation, Throwable thrown) {
+ 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.
*
}
/**
- * Logs a response. If the response is not of type, String, then it attempts to
+ * Logs a message. If the message is not of type, String, then it attempts to
* pretty-print it into JSON before logging.
*
* @param direction IN or OUT
* @param infra communication infrastructure on which it was published
* @param source source name (e.g., the URL or Topic name)
- * @param response response to be logged
+ * @param message message to be logged
* @return the JSON text that was logged
*/
- public <T> String logMessage(EventType direction, CommInfrastructure infra, String source, T response) {
+ public <T> String logMessage(EventType direction, CommInfrastructure infra, String source, T message) {
String json;
try {
- if (response == null) {
- json = null;
- } else if (response instanceof String) {
- json = response.toString();
- } else {
- json = makeCoder().encode(response, true);
- }
+ json = prettyPrint(message);
- } catch (CoderException e) {
+ } catch (IllegalArgumentException e) {
String type = (direction == EventType.IN ? "response" : "request");
logger.warn("cannot pretty-print {}", type, e);
- json = response.toString();
+ json = message.toString();
}
logger.info("[{}|{}|{}|]{}{}", direction, infra, source, NetLoggerUtil.SYSTEM_LS, json);
return json;
}
+ /**
+ * Converts a message to a "pretty-printed" String using the operation's normal
+ * serialization provider (i.e., it's <i>coder</i>).
+ *
+ * @param message response to be logged
+ * @return the JSON text that was logged
+ * @throws IllegalArgumentException if the message cannot be converted
+ */
+ public <T> String prettyPrint(T message) {
+ if (message == null) {
+ return null;
+ } else if (message instanceof String) {
+ return message.toString();
+ } else {
+ try {
+ return getCoder().encode(message, true);
+ } catch (CoderException e) {
+ throw new IllegalArgumentException("cannot encode message", e);
+ }
+ }
+ }
+
// these may be overridden by subclasses or junit tests
/**
// these may be overridden by junit tests
- protected Coder makeCoder() {
+ protected Coder getCoder() {
return coder;
}
}
Code Review / policy / models.git / blobdiff