2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
6 * Modifications Copyright (C) 2023-2024 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
12 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
22 package org.onap.policy.controlloop.eventmanager;
24 import java.util.ArrayDeque;
25 import java.util.Deque;
26 import java.util.LinkedHashMap;
28 import java.util.UUID;
29 import lombok.AccessLevel;
31 import lombok.NonNull;
33 import lombok.ToString;
34 import org.drools.core.WorkingMemory;
35 import org.kie.api.runtime.rule.FactHandle;
36 import org.onap.policy.controlloop.ControlLoopException;
37 import org.onap.policy.controlloop.actorserviceprovider.OperationFinalResult;
38 import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
39 import org.onap.policy.controlloop.actorserviceprovider.OperationResult;
40 import org.onap.policy.controlloop.actorserviceprovider.TargetType;
41 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
42 import org.onap.policy.controlloop.drl.legacy.ControlLoopParams;
43 import org.onap.policy.drools.domain.models.operational.Operation;
44 import org.onap.policy.drools.domain.models.operational.OperationalTarget;
45 import org.onap.policy.drools.system.PolicyEngine;
46 import org.onap.policy.drools.system.PolicyEngineConstants;
47 import org.slf4j.Logger;
48 import org.slf4j.LoggerFactory;
51 * Manager for a single control loop event. Processing progresses through each policy,
52 * which involves at least one step. As a step is processed, additional preprocessor steps
53 * may be pushed onto the queue (e.g., locks, A&AI queries, guards).
55 @ToString(onlyExplicitlyIncluded = true)
56 public abstract class ClEventManagerWithSteps<T extends Step> extends ControlLoopEventManager implements StepContext {
58 private static final Logger logger = LoggerFactory.getLogger(ClEventManagerWithSteps.class);
59 private static final long serialVersionUID = -1216568161322872641L;
62 * Maximum number of steps, for a single policy, allowed in the queue at a time. This
63 * prevents an infinite loop occurring with calls to {@link #loadPreprocessorSteps()}.
65 public static final int MAX_STEPS = 30;
68 LOAD_POLICY, POLICY_LOADED, AWAITING_OUTCOME, DONE
72 * Request ID, as a String.
75 private final String requestIdStr;
82 * {@code True} if the event has been accepted (i.e., an "ACTIVE" notification has
83 * been delivered), {@code false} otherwise.
87 private boolean accepted;
90 * Queue of steps waiting to be performed.
93 private final transient Deque<T> steps = new ArrayDeque<>(6);
96 * Policy currently being processed.
98 @Getter(AccessLevel.PROTECTED)
99 private Operation policy;
102 * Result of the last policy operation. This is just a place where the rules can store
103 * the value for passing to {@link #loadNextPolicy(OperationResult)}.
107 private OperationResult result = OperationResult.SUCCESS;
111 private int numOnsets = 1;
114 private int numAbatements = 0;
117 private OperationFinalResult finalResult = null;
120 * Message to be placed into the final notification. Typically used when something
121 * causes processing to abort.
124 private String finalMessage = null;
126 private final transient WorkingMemory workMem;
127 private transient FactHandle factHandle;
131 * Constructs the object.
133 * @param services services the manager should use when processing the event
134 * @param params control loop parameters
135 * @param requestId event request ID
136 * @param workMem working memory to update if this changes
137 * @throws ControlLoopException if the event is invalid or if a YAML processor cannot
140 protected ClEventManagerWithSteps(EventManagerServices services, ControlLoopParams params, UUID requestId,
141 WorkingMemory workMem) throws ControlLoopException {
143 super(services, params, requestId);
145 if (requestId == null) {
146 throw new ControlLoopException("No request ID");
149 this.workMem = workMem;
150 this.requestIdStr = getRequestId().toString();
154 public void destroy() {
155 for (T step : getSteps()) {
163 * Starts the manager and loads the first policy.
165 * @throws ControlLoopException if the processor cannot get a policy
167 public void start() throws ControlLoopException {
169 throw new IllegalStateException("manager is no longer active");
172 if ((factHandle = workMem.getFactHandle(this)) == null) {
173 throw new IllegalStateException("manager is not in working memory");
176 if (!getSteps().isEmpty()) {
177 throw new IllegalStateException("manager already started");
184 * Indicates that processing has been aborted.
186 * @param finalState final state
187 * @param finalResult final result
188 * @param finalMessage final message
190 public void abort(@NonNull State finalState, OperationFinalResult finalResult, String finalMessage) {
191 this.state = finalState;
192 this.finalResult = finalResult;
193 this.finalMessage = finalMessage;
197 * Loads the next policy.
199 * @param lastResult result from the last policy
201 * @throws ControlLoopException if the processor cannot get a policy
203 public void loadNextPolicy(@NonNull OperationResult lastResult) throws ControlLoopException {
204 getProcessor().nextPolicyForResult(lastResult);
209 * Loads the current policy.
211 * @throws ControlLoopException if the processor cannot get a policy
213 protected void loadPolicy() throws ControlLoopException {
214 if ((finalResult = getProcessor().checkIsCurrentPolicyFinal()) != null) {
215 // final policy - nothing more to do
219 policy = getProcessor().getCurrentPolicy();
221 var actor = policy.getActorOperation();
223 OperationalTarget target = actor.getTarget();
224 String targetType = (target != null ? target.getTargetType() : null);
225 Map<String, String> entityIds = (target != null ? target.getEntityIds() : null);
227 // convert policy payload from Map<String,String> to Map<String,Object>
228 Map<String, Object> payload = new LinkedHashMap<>();
229 if (actor.getPayload() != null) {
230 payload.putAll(actor.getPayload());
234 ControlLoopOperationParams params = ControlLoopOperationParams.builder()
235 .actorService(getActorService())
236 .actor(actor.getActor())
237 .operation(actor.getOperation())
238 .requestId(getRequestId())
239 .executor(getExecutor())
240 .retry(policy.getRetries())
241 .timeoutSec(policy.getTimeout())
242 .targetType(TargetType.toTargetType(targetType))
243 .targetEntityIds(entityIds)
245 .startCallback(this::onStart)
246 .completeCallback(this::onComplete)
250 // load the policy's operation
251 loadPolicyStep(params);
255 * Makes the step associated with the given parameters.
257 * @param params operation's parameters
259 protected abstract void loadPolicyStep(ControlLoopOperationParams params);
262 * Loads the preprocessor steps needed by the step that's at the front of the queue.
264 public void loadPreprocessorSteps() {
265 if (getSteps().size() >= MAX_STEPS) {
266 throw new IllegalStateException("too many steps");
269 // initialize the step so we can query its properties
270 assert getSteps().peek() != null;
271 getSteps().peek().init();
275 * Executes the first step in the queue.
277 * @return {@code true} if the step was started, {@code false} if it is no longer
278 * needed (or if the queue is empty)
280 public boolean executeStep() {
281 T step = getSteps().peek();
286 return step.start(getEndTimeMs() - System.currentTimeMillis());
290 * Discards the current step, if any.
292 public void nextStep() {
297 * Delivers a notification to a topic.
299 * @param sinkName name of the topic sink
300 * @param notification notification to be published, or {@code null} if nothing is to
302 * @param notificationType type of notification, used when logging error messages
303 * @param ruleName name of the rule doing the publishing
305 public <N> void deliver(String sinkName, N notification, String notificationType, String ruleName) {
307 if (notification != null) {
308 getPolicyEngineManager().deliver(sinkName, notification);
311 } catch (RuntimeException e) {
312 logger.warn("{}: {}.{}: manager={} exception publishing {}", getClosedLoopControlName(), getPolicyName(),
313 ruleName, this, notificationType, e);
317 protected int bumpOffsets() {
321 protected int bumpAbatements() {
322 return numAbatements++;
326 public void onStart(OperationOutcome outcome) {
327 super.onStart(outcome);
328 workMem.update(factHandle, this);
332 public void onComplete(OperationOutcome outcome) {
333 super.onComplete(outcome);
334 workMem.update(factHandle, this);
337 // these following methods may be overridden by junit tests
339 protected PolicyEngine getPolicyEngineManager() {
340 return PolicyEngineConstants.getManager();