From 56efff004af2d1be64c67f7c8091cb4553a0e86b Mon Sep 17 00:00:00 2001 From: Jim Hahn Date: Thu, 20 Aug 2020 10:11:54 -0400 Subject: Add generic eventmanager classes Added classes that are event-agnostic and support moving control from java into rules. Updates per review comments: - removed policy scope Issue-ID: POLICY-2748-event-mgr Change-Id: Icf811cc25a3975543fc5c725766b7b9df2bb87b0 Signed-off-by: Jim Hahn --- .../controlloop/eventmanager/ActorConstants.java | 34 +++ .../eventmanager/ControlLoopEventManager.java | 297 +++++++++++++++++++++ .../onap/policy/controlloop/eventmanager/Step.java | 253 ++++++++++++++++++ .../controlloop/eventmanager/StepContext.java | 72 +++++ 4 files changed, 656 insertions(+) create mode 100644 controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ActorConstants.java create mode 100644 controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ControlLoopEventManager.java create mode 100644 controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/Step.java create mode 100644 controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/StepContext.java (limited to 'controlloop/common/eventmanager/src/main') diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ActorConstants.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ActorConstants.java new file mode 100644 index 000000000..2591a3fa0 --- /dev/null +++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ActorConstants.java @@ -0,0 +1,34 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2020 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.controlloop.eventmanager; + +public class ActorConstants { + public static final String CL_TIMEOUT_ACTOR = "-CL-TIMEOUT-"; + public static final String LOCK_ACTOR = "LOCK"; + public static final String LOCK_OPERATION = "Lock"; + + public static final String PAYLOAD_KEY_VF_COUNT = "vfCount"; + + + private ActorConstants() { + // do nothing + } +} diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ControlLoopEventManager.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ControlLoopEventManager.java new file mode 100644 index 000000000..5cf087abd --- /dev/null +++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/ControlLoopEventManager.java @@ -0,0 +1,297 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2017-2020 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.controlloop.eventmanager; + +import java.io.Serializable; +import java.util.Deque; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.ToString; +import org.onap.policy.controlloop.ControlLoopException; +import org.onap.policy.controlloop.actorserviceprovider.ActorService; +import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome; +import org.onap.policy.controlloop.drl.legacy.ControlLoopParams; +import org.onap.policy.controlloop.ophistory.OperationHistoryDataManager; +import org.onap.policy.controlloop.processor.ControlLoopProcessor; +import org.onap.policy.drools.core.lock.LockCallback; +import org.onap.policy.drools.system.PolicyEngineConstants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Manager for a single event. Once this has been created, the event can be retracted from + * working memory. Invoke {@link #isActive()} to determine if the manager is active (i.e., + * hasn't been replicated from another server). When the manager is no longer needed, + * {@link #destroy()} should be invoked. + */ +@ToString(onlyExplicitlyIncluded = true) +public class ControlLoopEventManager implements StepContext, Serializable { + + private static final Logger logger = LoggerFactory.getLogger(ControlLoopEventManager.class); + private static final long serialVersionUID = -1216568161322872641L; + + private static final String EVENT_MANAGER_SERVICE_CONFIG = "event-manager"; + + /** + * Counts the number of these objects that have been created. This is used by junit + * tests. + */ + private static final AtomicLong createCount = new AtomicLong(0); + + /** + * {@code True} if this object was created by this JVM instance, {@code false} + * otherwise. This will be {@code false} if this object is reconstituted from a + * persistent store or by transfer from another server. + */ + private transient boolean createdByThisJvmInstance; + + @Getter + @ToString.Include + public final String closedLoopControlName; + @Getter + @ToString.Include + private final UUID requestId; + + /** + * Time, in milliseconds, when the control loop will time out. + */ + @Getter + private final long endTimeMs; + + // fields extracted from the ControlLoopParams + @Getter + private final String policyName; + @Getter + private final String policyVersion; + + /** + * Maps a target entity to its lock. + */ + private final transient Map target2lock = new HashMap<>(); + + @Getter(AccessLevel.PROTECTED) + private final ControlLoopProcessor processor; + + /** + * Set of properties used while processing the event. + */ + private Map properties = new ConcurrentHashMap<>(); + + /** + * Unprocessed outcomes from the operations. Outcomes are added to this each time the + * "start" or "complete" callback is invoked, typically by an operation running in a + * background thread, thus it must be thread safe. + */ + @Getter + private final transient Deque outcomes = new ConcurrentLinkedDeque<>(); + + + /** + * Constructs the object. + * + * @param params control loop parameters + * @param requestId event request ID + * @throws ControlLoopException if the event is invalid or if a YAML processor cannot + * be created + */ + public ControlLoopEventManager(ControlLoopParams params, UUID requestId) throws ControlLoopException { + + createCount.incrementAndGet(); + + this.createdByThisJvmInstance = true; + this.closedLoopControlName = params.getClosedLoopControlName(); + this.requestId = requestId; + this.policyName = params.getPolicyName(); + this.policyVersion = params.getPolicyVersion(); + this.processor = new ControlLoopProcessor(params.getToscaPolicy()); + this.endTimeMs = System.currentTimeMillis() + detmControlLoopTimeoutMs(); + } + + /** + * Gets the number of manager objects that have been created. + * + * @return the number of manager objects that have been created + */ + public static long getCreateCount() { + return createCount.get(); + } + + /** + * Determines if the manager is still active. + * + * @return {@code true} if the manager is still active, {@code false} otherwise + */ + public boolean isActive() { + return createdByThisJvmInstance; + } + + /** + * Cancels the current operation and frees all locks. + */ + public void destroy() { + if (isActive()) { + getBlockingExecutor().execute(this::freeAllLocks); + } + } + + /** + * Frees all locks. + */ + private void freeAllLocks() { + target2lock.values().forEach(LockData::free); + } + + /** + * Determines the overall control loop timeout. + * + * @return the policy timeout, in milliseconds, if specified, a default timeout + * otherwise + */ + private long detmControlLoopTimeoutMs() { + // validation checks preclude null or 0 timeout values in the policy + Integer timeout = processor.getControlLoop().getTimeout(); + return TimeUnit.MILLISECONDS.convert(timeout, TimeUnit.SECONDS); + } + + /** + * Requests a lock. This requests the lock for the time that remains before the + * timeout expires. This avoids having to extend the lock. + * + * @param targetEntity entity to be locked + * @return a future that can be used to await the lock + */ + @Override + public synchronized CompletableFuture requestLock(String targetEntity) { + + long remainingMs = endTimeMs - System.currentTimeMillis(); + int remainingSec = 15 + Math.max(0, (int) TimeUnit.SECONDS.convert(remainingMs, TimeUnit.MILLISECONDS)); + + LockData data = target2lock.computeIfAbsent(targetEntity, key -> { + LockData data2 = new LockData(key, requestId); + makeLock(targetEntity, requestId.toString(), remainingSec, data2); + + data2.addUnavailableCallback(this::onComplete); + + return data2; + }); + + return data.getFuture(); + } + + public void onStart(OperationOutcome outcome) { + outcomes.add(outcome); + } + + public void onComplete(OperationOutcome outcome) { + outcomes.add(outcome); + } + + /** + * Determines if the context contains a property. + * + * @param name name of the property of interest + * @return {@code true} if the context contains the property, {@code false} otherwise + */ + public boolean contains(String name) { + return properties.containsKey(name); + } + + /** + * Gets a property, casting it to the desired type. + * + * @param desired type + * @param name name of the property whose value is to be retrieved + * @return the property's value, or {@code null} if it does not yet have a value + */ + @SuppressWarnings("unchecked") + public T getProperty(String name) { + return (T) properties.get(name); + } + + /** + * Sets a property's value. + * + * @param name property name + * @param value new property value + */ + public void setProperty(String name, Serializable value) { + logger.error("set property {}={} manager={}", name, value, this); + properties.put(name, value); + } + + /** + * Removes a property. + * + * @param name property name + */ + public void removeProperty(String name) { + properties.remove(name); + } + + /** + * Initializes various components, on demand. + */ + private static class LazyInitData { + private static final OperationHistoryDataManager DATA_MANAGER; + private static final ActorService ACTOR_SERVICE; + + static { + // TODO how to dynamically change data manager, depending whether or not + // guards are enabled? + EventManagerServices services = new EventManagerServices(EVENT_MANAGER_SERVICE_CONFIG); + ACTOR_SERVICE = services.getActorService(); + DATA_MANAGER = services.getDataManager(); + } + } + + // the following methods may be overridden by junit tests + + public Executor getExecutor() { + return ForkJoinPool.commonPool(); + } + + protected ExecutorService getBlockingExecutor() { + return PolicyEngineConstants.getManager().getExecutorService(); + } + + protected void makeLock(String targetEntity, String requestId, int holdSec, LockCallback callback) { + PolicyEngineConstants.getManager().createLock(targetEntity, requestId, holdSec, callback, false); + } + + public ActorService getActorService() { + return LazyInitData.ACTOR_SERVICE; + } + + public OperationHistoryDataManager getDataManager() { + return LazyInitData.DATA_MANAGER; + } +} diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/Step.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/Step.java new file mode 100644 index 000000000..01c64e5bc --- /dev/null +++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/Step.java @@ -0,0 +1,253 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2020 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.controlloop.eventmanager; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import lombok.Getter; +import lombok.NonNull; +import org.onap.policy.controlloop.actorserviceprovider.Operation; +import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome; +import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams; +import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A single step within a single policy. The rules typically break a policy operation down + * into separate steps. For instance, a policy for VF-Module-Create would be broken down + * into lock target, A&AI tenant query, A&AI custom query, guard, and finally + * VF-Module-Create. + */ +public class Step { + private static final Logger logger = LoggerFactory.getLogger(Step.class); + + @Getter + protected ControlLoopOperationParams params; + + /** + * Time when the first step started for the current policy. This is shared by all + * steps for the policy. When a step is started and finds this to be {@code null}, it + * sets the value. Subsequent steps leave it unchanged. + */ + private final AtomicReference startTime; + + /** + * {@code True} if this step is for the policy's actual operation, {@code false} if it's a preprocessor step. + */ + @Getter + private final boolean policyStep; + + /** + * The operation for this step. + */ + @Getter + private Operation operation = null; + + /** + * Used to cancel the running operation. + */ + protected CompletableFuture future; + + + /** + * Constructs the object. This is used when constructing the step for the policy's actual operation. + * + * @param params operation parameters + * @param startTime start time of the first step for the current policy, initially + * containing {@code null} + */ + public Step(ControlLoopOperationParams params, @NonNull AtomicReference startTime) { + this.params = params; + this.startTime = startTime; + this.policyStep = true; + } + + /** + * Constructs the object using information from another step. This is used when constructing a preprocessing + * step. + * + * @param otherStep step whose information should be used + * @param actor actor name + * @param operation operation name + */ + public Step(Step otherStep, String actor, String operation) { + this.params = otherStep.params.toBuilder().actor(actor).operation(operation).retry(null).timeoutSec(null) + .payload(new LinkedHashMap<>()).build(); + this.startTime = otherStep.startTime; + this.policyStep = false; + } + + public String getActorName() { + return params.getActor(); + } + + public String getOperationName() { + return params.getOperation(); + } + + /** + * Determines if the operation has been initialized (i.e., created). + * + * @return {@code true} if the operation has been initialized, {@code false} otherwise + */ + public boolean isInitialized() { + return (operation != null); + } + + /** + * Initializes the step, creating the operation if it has not yet been created. + */ + public void init() { + if (operation == null) { + operation = buildOperation(); + } + } + + /** + * Starts the operation. + * + * @param remainingMs time remaining, in milliseconds, for the control loop + * @return {@code true} if started, {@code false} if the step is no longer necessary + * (i.e., because it was previously completed) + */ + public boolean start(long remainingMs) { + if (!isInitialized()) { + throw new IllegalStateException("step has not been initialized"); + } + + if (future != null) { + throw new IllegalStateException("step is already running"); + } + + try { + initStartTime(); + future = operation.start(); + + // handle any exceptions that may be thrown, set timeout, and handle timeout + + // @formatter:off + future.exceptionally(this::handleException) + .orTimeout(remainingMs, TimeUnit.MILLISECONDS) + .exceptionally(this::handleTimeout); + // @formatter:on + + } catch (RuntimeException e) { + handleException(e); + } + + return true; + } + + /** + * Handles exceptions that may be generated. + * + * @param thrown exception that was generated + * @return {@code null} + */ + private OperationOutcome handleException(Throwable thrown) { + if (thrown instanceof CancellationException || thrown.getCause() instanceof CancellationException) { + return null; + } + + logger.warn("{}.{}: exception starting operation for {}", params.getActor(), params.getOperation(), + params.getRequestId(), thrown); + OperationOutcome outcome = new PipelineUtil(params).setOutcome(params.makeOutcome(), thrown); + outcome.setStart(startTime.get()); + outcome.setEnd(Instant.now()); + outcome.setFinalOutcome(true); + params.getCompleteCallback().accept(outcome); + + // this outcome is not used so just return "null" + return null; + } + + /** + * Handles control loop timeout exception. + * + * @param thrown exception that was generated + * @return {@code null} + */ + private OperationOutcome handleTimeout(Throwable thrown) { + logger.warn("{}.{}: control loop timeout for {}", params.getActor(), params.getOperation(), + params.getRequestId(), thrown); + + OperationOutcome outcome = new PipelineUtil(params).setOutcome(params.makeOutcome(), thrown); + outcome.setActor(ActorConstants.CL_TIMEOUT_ACTOR); + outcome.setOperation(null); + outcome.setStart(startTime.get()); + outcome.setEnd(Instant.now()); + outcome.setFinalOutcome(true); + params.getCompleteCallback().accept(outcome); + + // cancel the operation, if it's still running + future.cancel(false); + + // this outcome is not used so just return "null" + return null; + } + + /** + * Cancels the operation, if it's running. + */ + public void cancel() { + if (future != null) { + future.cancel(false); + } + } + + /** + * Initializes the start time, if it's still unset. + */ + private void initStartTime() { + if (startTime.get() == null) { + startTime.set(Instant.now()); + } + } + + /** + * Gets the start time. + * + * @return the start time, or {@code null} if it hasn't been initialized yet + */ + public Instant getStartTime() { + return startTime.get(); + } + + /** + * Builds the operation. The default method simply invokes + * {@link ControlLoopOperationParams#build()}. + * + * @return a new operation + */ + protected Operation buildOperation() { + return params.build(); + } + + @Override + public String toString() { + return "Step(actor=" + getActorName() + ", operation=" + getOperationName() + ")"; + } +} diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/StepContext.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/StepContext.java new file mode 100644 index 000000000..5251b7acc --- /dev/null +++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/controlloop/eventmanager/StepContext.java @@ -0,0 +1,72 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2020 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.controlloop.eventmanager; + +import java.io.Serializable; +import java.util.concurrent.CompletableFuture; +import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome; + +/** + * Context used by steps to perform their work. + */ +public interface StepContext { + + /** + * Determines if the context contains a property. + * + * @param name name of the property of interest + * @return {@code true} if the context contains the property, {@code false} otherwise + */ + public boolean contains(String name); + + /** + * Gets a property, casting it to the desired type. + * + * @param desired type + * @param name name of the property whose value is to be retrieved + * @return the property's value, or {@code null} if it does not yet have a value + */ + public T getProperty(String name); + + /** + * Sets a property's value. + * + * @param name property name + * @param value new property value + */ + public void setProperty(String name, Serializable value); + + /** + * Removes a property. + * + * @param name property name + */ + public void removeProperty(String name); + + /** + * Requests a lock. This requests the lock for the time that remains before the + * timeout expires. This avoids having to extend the lock. + * + * @param targetEntity entity to be locked + * @return a future that can be used to await the lock + */ + public CompletableFuture requestLock(String targetEntity); +} -- cgit 1.2.3-korg