diff options
Diffstat (limited to 'appc-oam/appc-oam-bundle/src/main')
10 files changed, 472 insertions, 192 deletions
diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseActionRunnable.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseActionRunnable.java index 542e53d63..b118eb0d1 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseActionRunnable.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseActionRunnable.java @@ -45,9 +45,6 @@ import java.util.concurrent.Future; * <br> - finalState */ public abstract class BaseActionRunnable extends BaseCommon implements Runnable { - final String OAM_OPERATION_TIMEOUT_SECOND = "appc.OAM.api.timeout"; - /** Default operation tiemout set to 1 minute */ - final int DEFAULT_OAM_OPERATION_TIMEOUT = 60; /** Abort due to rejection message format with flexible operation name */ final String ABORT_MESSAGE_FORMAT = "Aborting %s operation due to %s."; /** Timeout message format with flexible operation name */ @@ -58,7 +55,6 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable final String DUE_TO_EXECUTION_ERROR = "due to execution error."; private boolean isWaiting = false; - private AppcOamStates currentState; long startTimeMs = 0; long timeoutMs = 0; boolean doTimeoutChecking = false; @@ -80,32 +76,25 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable rpc = parent.rpc; commonHeader = parent.commonHeader; - startTime = parent.startTime; myParent = parent; - setTimeoutValues(); } /** - * Set timeout in milliseconds + * Collect the timeout value for this {@link BaseActionRunnable} */ void setTimeoutValues() { - Integer timeoutSeconds = myParent.timeoutSeconds; - if (timeoutSeconds == null) { - timeoutMs = configurationHelper.getConfig().getIntegerProperty( - OAM_OPERATION_TIMEOUT_SECOND, DEFAULT_OAM_OPERATION_TIMEOUT) * 1000; - } else { - timeoutMs = timeoutSeconds.longValue() * 1000; - } - + startTime = myParent.startTime; + timeoutMs = myParent.getTimeoutMilliseconds(); doTimeoutChecking = timeoutMs != 0; if (doTimeoutChecking) { startTimeMs = startTime.getTime(); } logDebug("%s action runnable check timeout (%s) with timeout (%d)ms, and startMs (%d)", - rpc.name(), Boolean.toString(doTimeoutChecking), timeoutMs, startTimeMs); + rpc.name(), Boolean.toString(doTimeoutChecking), timeoutMs, startTimeMs); } + /** * Abort operation handling due to outside interruption, does<br> * - set ABORT status<br> @@ -114,7 +103,7 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable * * @param newRpc of the new AppcOam.RPC operation. */ - public void abortRunnable(final AppcOam.RPC newRpc) { + void abortRunnable(final AppcOam.RPC newRpc) { resetLogProperties(false); String additionalMsg = String.format(NEW_RPC_OPERATION_REQUEST, newRpc); @@ -131,7 +120,7 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable try { setInitialLogProperties(); logDebug(String.format("===========in %s run (waiting: %s)=======", - actionName, Boolean.toString(isWaiting))); + actionName, Boolean.toString(isWaiting))); if (isWaiting) { if (!checkState()) { @@ -159,7 +148,7 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable */ void keepWaiting() { logDebug(String.format("%s runnable waiting, current state is %s.", - actionName, currentState == null ? "null" : currentState.toString())); + actionName, stateHelper.getCurrentOamState())); isTimeout("keepWaiting"); } @@ -173,9 +162,9 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable boolean isTimeout(String parentName) { logDebug(String.format("%s task isTimeout called from %s", actionName, parentName)); if (doTimeoutChecking - && System.currentTimeMillis() - startTimeMs > timeoutMs) { + && System.currentTimeMillis() - startTimeMs > timeoutMs) { logger.error(String.format("%s operation timeout (%d) ms has reached, abort with error state.", - actionName, timeoutMs)); + actionName, timeoutMs)); setStatus(OAMCommandStatus.TIMEOUT, String.format(TIMEOUT_MESSAGE_FORMAT, rpc.name(), timeoutMs)); postAction(AppcOamStates.Error); @@ -253,8 +242,7 @@ public abstract class BaseActionRunnable extends BaseCommon implements Runnable return true; } - currentState = stateHelper.getBundlesState(); - if (currentState == finalState) { + if (stateHelper.getBundlesState() == finalState) { setStatus(OAMCommandStatus.SUCCESS); postDoAction(true); return true; diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseCommon.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseCommon.java index ccb57305a..9c28007b4 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseCommon.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseCommon.java @@ -53,7 +53,6 @@ import static com.att.eelf.configuration.Configuration.MDC_INSTANCE_UUID; import static com.att.eelf.configuration.Configuration.MDC_KEY_REQUEST_ID; import static com.att.eelf.configuration.Configuration.MDC_SERVER_FQDN; import static com.att.eelf.configuration.Configuration.MDC_SERVER_IP_ADDRESS; -import static com.att.eelf.configuration.Configuration.MDC_SERVICE_INSTANCE_ID; import static com.att.eelf.configuration.Configuration.MDC_SERVICE_NAME; /** @@ -61,7 +60,7 @@ import static com.att.eelf.configuration.Configuration.MDC_SERVICE_NAME; * - BaseProcessor (for REST sync handling) <br> * - BaseActionRunnable (for REST async handling) */ -abstract class BaseCommon { +public abstract class BaseCommon { final EELFLogger logger; final ConfigurationHelper configurationHelper; final StateHelper stateHelper; @@ -74,15 +73,14 @@ abstract class BaseCommon { CommonHeader commonHeader; private final List<String> MDC_KEYS = Arrays.asList( - LoggingConstants.MDCKeys.PARTNER_NAME, - LoggingConstants.MDCKeys.SERVER_NAME, - MDC_INSTANCE_UUID, - MDC_KEY_REQUEST_ID, - MDC_SERVER_FQDN, - MDC_SERVER_IP_ADDRESS, - MDC_SERVICE_NAME + LoggingConstants.MDCKeys.PARTNER_NAME, + LoggingConstants.MDCKeys.SERVER_NAME, + MDC_INSTANCE_UUID, + MDC_KEY_REQUEST_ID, + MDC_SERVER_FQDN, + MDC_SERVER_IP_ADDRESS, + MDC_SERVICE_NAME ); - private Map<String, String> oldMdcContent = new HashMap<>(); /** @@ -110,19 +108,19 @@ abstract class BaseCommon { void auditInfoLog(Msg msg) { LoggingUtils.auditInfo(startTime.toInstant(), new Date(System.currentTimeMillis()).toInstant(), - String.valueOf(status.getCode()), - status.getMessage(), - getClass().getCanonicalName(), - msg, - configurationHelper.getAppcName(), - stateHelper.getCurrentOamState().toString() + String.valueOf(status.getCode()), + status.getMessage(), + getClass().getCanonicalName(), + msg, + configurationHelper.getAppcName(), + stateHelper.getCurrentOamState().toString() ); } /** * Set MDC properties. */ - void setInitialLogProperties() { + public final void setInitialLogProperties() { MDC.put(MDC_KEY_REQUEST_ID, commonHeader.getRequestId()); MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, commonHeader.getOriginatorId()); MDC.put(MDC_INSTANCE_UUID, ""); // value should be created in the future @@ -141,15 +139,14 @@ abstract class BaseCommon { /** * Clear MDC properties. */ - void clearRequestLogProperties() { - try { - MDC.remove(MDC_KEY_REQUEST_ID); - MDC.remove(MDC_SERVICE_INSTANCE_ID); - MDC.remove(MDC_SERVICE_NAME); - MDC.remove(LoggingConstants.MDCKeys.PARTNER_NAME); - MDC.remove(LoggingConstants.MDCKeys.TARGET_VIRTUAL_ENTITY); - } catch (Exception e) { - logger.error("Unable to clear the Request Log properties" + e.getMessage()); + public final void clearRequestLogProperties() { + for (String key : MDC_KEYS) { + try { + MDC.remove(key); + } catch (Exception e) { + logger.error( + String.format("Unable to clear the Log properties (%s) due to exception: %s", key, e.getMessage())); + } } } @@ -227,24 +224,24 @@ abstract class BaseCommon { errorMessage = EELFResourceManager.format(Msg.OAM_OPERATION_INVALID_INPUT, t.getMessage()); } else if (t instanceof InvalidStateException) { exceptionMessage = String.format(AppcOam.INVALID_STATE_MESSAGE_FORMAT, - rpc.getAppcOperation(), appName, stateHelper.getCurrentOamState()); + rpc.getAppcOperation(), appName, stateHelper.getCurrentOamState()); oamCommandStatus = OAMCommandStatus.REJECTED; errorMessage = EELFResourceManager.format(Msg.INVALID_STATE_TRANSITION, exceptionMessage); } else { oamCommandStatus = OAMCommandStatus.UNEXPECTED_ERROR; errorMessage = EELFResourceManager.format(Msg.OAM_OPERATION_EXCEPTION, t, - appName, t.getClass().getSimpleName(), rpc.name(), exceptionMessage); + appName, t.getClass().getSimpleName(), rpc.name(), exceptionMessage); } setStatus(oamCommandStatus, exceptionMessage); LoggingUtils.logErrorMessage( - String.valueOf(status.getCode()), - status.getMessage(), - LoggingConstants.TargetNames.APPC, - LoggingConstants.TargetNames.APPC_OAM_PROVIDER, - errorMessage, - AppcOam.class.getCanonicalName()); + String.valueOf(status.getCode()), + status.getMessage(), + LoggingConstants.TargetNames.APPC, + LoggingConstants.TargetNames.APPC_OAM_PROVIDER, + errorMessage, + AppcOam.class.getCanonicalName()); resetLogProperties(true); } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseProcessor.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseProcessor.java index 784beccc3..aa5423d3b 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseProcessor.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/BaseProcessor.java @@ -40,6 +40,8 @@ import org.openecomp.appc.statemachine.impl.readers.AppcOamStates; import java.util.Date; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; /** * Base processor for OAM APIs, such as maintenance mode, restart, start and stop API. @@ -48,11 +50,14 @@ import java.util.concurrent.Future; * <p>Specific API processor will overwrite the general methods to add specific behaviors. */ public abstract class BaseProcessor extends BaseCommon { + /** lock to serialize incoming OAM operations. */ + private static final Object LOCK = new Object(); + final AsyncTaskHelper asyncTaskHelper; final BundleHelper bundleHelper; - - Integer timeoutSeconds; + /** the requestTimeoutSeconds to use for this OAM operation */ + private Integer requestTimeoutSeconds; Msg auditMsg; BaseActionRunnable runnable; private Future<?> scheduledRunnable = null; @@ -90,7 +95,8 @@ public abstract class BaseProcessor extends BaseCommon { try { preProcess(requestInput); - timeoutSeconds = operationHelper.getParamRequestTimeout(requestInput); + //The OAM request may specify timeout value + requestTimeoutSeconds = operationHelper.getParamRequestTimeout(requestInput); scheduleAsyncTask(); } catch (Exception e) { setErrorStatus(e); @@ -112,13 +118,31 @@ public abstract class BaseProcessor extends BaseCommon { * @throws APPCException when state validation failed */ protected void preProcess(final Object requestInput) - throws InvalidInputException, APPCException, InvalidStateException { + throws InvalidInputException, APPCException, InvalidStateException,InterruptedException,TimeoutException { + setInitialLogProperties(); operationHelper.isInputValid(requestInput); - AppcOamStates nextState = operationHelper.getNextState( - rpc.getAppcOperation(), stateHelper.getCurrentOamState()); - setInitialLogProperties(); - stateHelper.setState(nextState); + //All OAM operation pass through here first to validate if an OAM state change is allowed. + //If a state change is allowed cancel the occurring OAM (if any) before starting this one. + //we will synchronized so that only one can do this at any given time. + synchronized(LOCK) { + AppcOamStates currentOamState = stateHelper.getCurrentOamState(); + + //make sure this OAM operation can transition to the desired OAM operation + AppcOamStates nextState = operationHelper.getNextState( + rpc.getAppcOperation(), currentOamState); + + stateHelper.setState(nextState); + + //cancel the BaseActionRunnable currently executing + //it got to be completely terminated before proceeding + asyncTaskHelper.cancelBaseActionRunnable( + rpc, + currentOamState, + getTimeoutMilliseconds(), + TimeUnit.MILLISECONDS + ); + } } /** @@ -135,32 +159,49 @@ public abstract class BaseProcessor extends BaseCommon { protected void scheduleAsyncTask() { if (runnable == null) { logger.error(String.format( - "Skipped schedule async task for rpc(%s) due to runnable is null", rpc.name())); + "Skipped schedule async task for rpc(%s) due to runnable is null", rpc.name())); return; } - scheduledRunnable = asyncTaskHelper.scheduleAsyncTask(rpc, runnable); + scheduledRunnable = asyncTaskHelper.scheduleBaseRunnable( + runnable, runnable::abortRunnable, getInitialDelayMillis(), getDelayMillis()); } + /** - * Check if current running task is the same as schedule task - * @return true if they are the same, otherwise false. + * The timeout for this OAM operation. The timeout source is chosen in the following order: + * request, config file, default value + * @return - the timeout for this OAM operation. */ - boolean isSameAsyncTask() { - return asyncTaskHelper.getCurrentAsyncTask() == scheduledRunnable; + long getTimeoutMilliseconds() { + return configurationHelper.getOAMOperationTimeoutValue(this.requestTimeoutSeconds); } + /** - * Cancel schedueled async task through AsyncTaskHelper + * @return initialDelayMillis - the time to delay first execution of {@link BaseActionRunnable} + */ + protected long getInitialDelayMillis(){ + return 0L; + } + + /** + * @return delayMillis the delay between the consecutive executions of {@link BaseActionRunnable} + */ + private long getDelayMillis(){ + return 1000L; + } + + /** + * Cancel the scheduled {@link BaseActionRunnable} through AsyncTaskHelper */ void cancelAsyncTask() { if (scheduledRunnable == null) { logger.error(String.format( - "Skipped cancel schedule async task for rpc(%s) due to scheduledRunnable is null", rpc.name())); + "Skipped cancel schedule async task for rpc(%s) due to scheduledRunnable is null", rpc.name())); return; } - - asyncTaskHelper.cancelAsyncTask(scheduledRunnable); - scheduledRunnable = null; + scheduledRunnable.cancel(true); } + } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamMmodeProcessor.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamMmodeProcessor.java index 973d0af36..6a0466022 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamMmodeProcessor.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamMmodeProcessor.java @@ -39,6 +39,8 @@ import org.openecomp.appc.requesthandler.LCMStateManager; import org.openecomp.appc.requesthandler.RequestHandler; import org.openecomp.appc.statemachine.impl.readers.AppcOamStates; +import java.util.concurrent.TimeoutException; + /** * Processor to handle maintenance mode OAM API. */ @@ -65,7 +67,7 @@ public class OamMmodeProcessor extends BaseProcessor { @Override protected void preProcess(final Object requestInput) - throws InvalidInputException, InvalidStateException, APPCException { + throws InvalidInputException, InvalidStateException, APPCException, InterruptedException, TimeoutException { super.preProcess(requestInput); //Close the gate so that no more new LCM request will be excepted. @@ -80,14 +82,25 @@ public class OamMmodeProcessor extends BaseProcessor { } /** + * {@inheritDoc} + * For maintenance mode we want a longer delay before initial execution of {@link BaseActionRunnable} + * so that any accepted LCM actions have time to git scheduled in the Dispatcher. + */ + @Override + protected long getInitialDelayMillis(){ + //wait ten seconds before + return 10000L; + } + + /** * This runnable does the async handling for the maintenance mode REST API, and will be scheduled to run * until terminating condition reaches. * - * <p>The runnable will conintue run if: <br> + * <p>The runnable will continue run if: <br> * - the runnable is not canceled outside <br> * - the in progress LCM request count is not zero<br> * <p> When LCM request count reaches to zero, this runnable will: <br> - * - post message through operatonHelper <br> + * - post message through operationHelper <br> * - set APP-C OAM state to maintenance mode <br> * - audit log the state <br> * - terminate this runnable itself <br> @@ -98,7 +111,7 @@ public class OamMmodeProcessor extends BaseProcessor { MyRunnable(BaseProcessor parent) { super(parent); - actionName = "OAM Maintanence mode"; + actionName = "OAM Maintenance mode"; auditMsg = Msg.OAM_OPERATION_MAINTENANCE_MODE; finalState = AppcOamStates.MaintenanceMode; } @@ -113,12 +126,6 @@ public class OamMmodeProcessor extends BaseProcessor { boolean checkState() { logDebug(String.format("Executing %s task", actionName)); - if (!myParent.isSameAsyncTask()) { - // cancel myself if I am not the current backgroundOamTask - myParent.cancelAsyncTask(); - logDebug(String.format("Finished %s task due to task removed", actionName)); - return true; - } boolean hasError = false; try { @@ -156,8 +163,8 @@ public class OamMmodeProcessor extends BaseProcessor { @Override void keepWaiting() { logDebug("The application '%s' has '%s' outstanding LCM request to complete" + - " before coming to a complete maintenance_mode.", - configurationHelper.getAppcName(), inprogressRequestCount); + " before coming to a complete maintenance_mode.", + configurationHelper.getAppcName(), inprogressRequestCount); } } } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamRestartProcessor.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamRestartProcessor.java index e9f0ada56..857578917 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamRestartProcessor.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamRestartProcessor.java @@ -112,14 +112,14 @@ public class OamRestartProcessor extends BaseProcessor { @Override boolean doAction() { logDebug(String.format("Executing %s task at phase (%s)", - actionName, currentPhase == null ? "null" : currentPhase.name())); + actionName, currentPhase == null ? "null" : currentPhase.name())); boolean isBundleOperationCompleted = true; try { switch (currentPhase) { case ToStop: isBundleOperationCompleted = bundleHelper.bundleOperations( - AppcOam.RPC.stop, bundleNameToFuture, myParent.asyncTaskHelper); + AppcOam.RPC.stop, bundleNameToFuture, myParent.asyncTaskHelper, this); currentPhase = ActionPhases.Stopped; break; case Stopped: @@ -129,12 +129,12 @@ public class OamRestartProcessor extends BaseProcessor { currentPhase = ActionPhases.ToStart; } else { logDebug(String.format("%s task is waiting in stopped phase, current state is %s", - actionName, currentState)); + actionName, currentState)); } break; case ToStart: isBundleOperationCompleted = bundleHelper.bundleOperations( - AppcOam.RPC.start, bundleNameToFuture, myParent.asyncTaskHelper); + AppcOam.RPC.start, bundleNameToFuture, myParent.asyncTaskHelper, this); currentPhase = ActionPhases.Started; break; case Error: @@ -143,13 +143,13 @@ public class OamRestartProcessor extends BaseProcessor { default: // Should not reach log it and return false; logger.error("%s task doAction reached %s phase. not supported. return false.", - actionName, currentPhase.name()); + actionName, currentPhase.name()); stateHelper.setState(AppcOamStates.Error); return false; } if (isTimeout("restart doAction") - || hasBundleOperationFailure()) { + || hasBundleOperationFailure()) { currentPhase = ActionPhases.Error; return true; } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStartProcessor.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStartProcessor.java index 0060bfcae..0d2f50513 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStartProcessor.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStartProcessor.java @@ -101,7 +101,7 @@ public class OamStartProcessor extends BaseProcessor { if (stateHelper.getState() != AppcOamStates.Started) { logDebug("Start - APPC OAM state is not started, start the bundles"); isBundleOperationCompleted = bundleHelper.bundleOperations( - rpc, bundleNameToFuture, myParent.asyncTaskHelper); + rpc, bundleNameToFuture, myParent.asyncTaskHelper, this); } if (isBundleOperationCompleted) { diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStopProcessor.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStopProcessor.java index d81901638..d8d88d319 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStopProcessor.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/processor/OamStopProcessor.java @@ -94,7 +94,7 @@ public class OamStopProcessor extends BaseProcessor { try { boolean isBundleOperationCompleted = bundleHelper.bundleOperations( - rpc, bundleNameToFuture, myParent.asyncTaskHelper); + rpc, bundleNameToFuture, myParent.asyncTaskHelper, this); if (isBundleOperationCompleted) { return true; } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/AsyncTaskHelper.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/AsyncTaskHelper.java index db6033752..0a4b868a8 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/AsyncTaskHelper.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/AsyncTaskHelper.java @@ -27,34 +27,61 @@ package org.openecomp.appc.oam.util; import com.att.eelf.configuration.EELFLogger; import org.openecomp.appc.oam.AppcOam; import org.openecomp.appc.oam.processor.BaseActionRunnable; +import org.openecomp.appc.statemachine.impl.readers.AppcOamStates; import org.osgi.framework.Bundle; import org.osgi.framework.FrameworkUtil; +import java.util.HashSet; +import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; /** - * Utility class provides general async task related help. + * The AsyncTaskHelper class manages an internal parent child data structure. The parent is a transient singleton, + * meaning only one can exist at any given time. The parent is scheduled with the + * {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)} and is executed at configured interval. It can be + * terminated by using the {@link Future#cancel(boolean)} or the {@link Future#cancel(boolean)} returned from \ + * {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)}. + * <p> + * The children are scheduled using {@link #submitBaseSubCallable(Callable)}} and can only be scheduled if a parent + * is scheduled. Children only execute once, but can be terminated preemptively by the {@link Future#cancel(boolean)} + * returned from {@link #submitBaseSubCallable(Callable)} or indirectly by terminating the parent via the method + * described above. + * <p> + * This class augments the meaning of {@link Future#isDone()} in that it guarantees that this method only returns true + * if the scheduled {@link Runnable} or {@link Callable} is not currently executing and is not going to execute in the + * future. This is different than the Java core implementation of {@link Future#isDone()} in which it will return + * true immediately after the {@link Future#cancel(boolean)} is called. Even if a Thread is actively executing the + * {@link Runnable} or {@link Callable} and has not return yet. See Java BUG JDK-8073704 + * <p> + * The parent {@link Future#isDone()} has an additional augmentation in that it will not return true until all of its + * children's {@link Future#isDone()} also return true. + * */ @SuppressWarnings("unchecked") public class AsyncTaskHelper { - final int MMODE_TASK_DELAY = 10000; - final int COMMON_INITIAL_DELAY = 0; - final int COMMON_INTERVAL = 1000; private final EELFLogger logger; private final ScheduledExecutorService scheduledExecutorService; private final ThreadPoolExecutor bundleOperationService; - /** Reference to the Async task */ - private volatile Future<?> backgroundOamTask; - /** Reference to the runnable of Async task */ - private volatile BaseActionRunnable taskRunnable; + /** Reference to {@link MyFuture} return from {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)} */ + private MyFuture backgroundBaseRunnableFuture; + + /** The cancel Callback from {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)} */ + private Consumer<AppcOam.RPC> cancelCallBackForBaseRunnable; + + /** All Futures created by thus calls which have not completed -- {@link Future#isDone()} equals false */ + private Set<MyFuture> myFutureSet = new HashSet<>(); /** * Constructor @@ -64,103 +91,162 @@ public class AsyncTaskHelper { logger = eelfLogger; scheduledExecutorService = Executors.newSingleThreadScheduledExecutor( - (runnable) -> { - Bundle bundle = FrameworkUtil.getBundle(AppcOam.class); - return new Thread(runnable, bundle.getSymbolicName() + " scheduledExecutor"); - } + (runnable) -> { + Bundle bundle = FrameworkUtil.getBundle(AppcOam.class); + return new Thread(runnable, bundle.getSymbolicName() + " scheduledExecutor"); + } ); bundleOperationService = new ThreadPoolExecutor( - 0, - 10, - 10, - TimeUnit.SECONDS, - new LinkedBlockingQueue(),// BlockingQueue<Runnable> workQueue - (runnable) -> new Thread(runnable, "OAM bundler operation executor")//ThreadFactory + 0, + 10, + 10, + TimeUnit.SECONDS, + new LinkedBlockingQueue(), //BlockingQueue<Runnable> workQueue + (runnable) -> { + Bundle bundle = FrameworkUtil.getBundle(AppcOam.class); + return new Thread(runnable, bundle.getSymbolicName() + " bundle operation executor"); + } ); } - void addThreadsToPool() { - bundleOperationService.setCorePoolSize(bundleOperationService.getMaximumPoolSize()); - } - - void removeThreadsFromPoolWhenDone() { - bundleOperationService.setCorePoolSize(0); - } - /** * Terminate the class <bS>ScheduledExecutorService</b> */ public void close() { logDebug("Start shutdown scheduleExcutorService."); - scheduledExecutorService.shutdown(); - bundleOperationService.shutdown(); + bundleOperationService.shutdownNow(); + scheduledExecutorService.shutdownNow(); logDebug("Completed shutdown scheduleExcutorService."); } + /** - * Get current async task refernce - * @return the class <b>backgroundOamTask</b> + * Cancel currently executing {@link BaseActionRunnable} if any. + * This method returns immediately if there is currently no {@link BaseActionRunnable} actively executing. + * @param rpcCausingAbort - The RPC causing the abort + * @param stateBeingAbborted - The current state being canceled + * @param timeout - The amount of time to wait for a cancel to complete + * @param timeUnit - The unit of time of timeout + * @throws TimeoutException - If {@link BaseActionRunnable} has not completely cancelled within the timeout period + * @throws InterruptedException - If the Thread waiting for the abort */ - public Future<?> getCurrentAsyncTask() { - return backgroundOamTask; + public synchronized void cancelBaseActionRunnable(final AppcOam.RPC rpcCausingAbort, + AppcOamStates stateBeingAbborted, + long timeout, TimeUnit timeUnit) + throws TimeoutException,InterruptedException { + + final MyFuture localBackgroundBaseRunnableFuture = backgroundBaseRunnableFuture; + final Consumer<AppcOam.RPC> localCancelCallBackForBaseRunnable = cancelCallBackForBaseRunnable; + + if (localBackgroundBaseRunnableFuture == null || localBackgroundBaseRunnableFuture.isDone()) { + return; + } + + if (localCancelCallBackForBaseRunnable != null) { + localCancelCallBackForBaseRunnable.accept(rpcCausingAbort); + } + localBackgroundBaseRunnableFuture.cancel(true); + + long timeoutMillis = timeUnit.toMillis(timeout); + long expiryTime = System.currentTimeMillis() + timeoutMillis; + while (!(localBackgroundBaseRunnableFuture.isDone())) { + long sleepTime = expiryTime - System.currentTimeMillis(); + if (sleepTime < 1) { + break; + } + this.wait(sleepTime); + } + + if (!localBackgroundBaseRunnableFuture.isDone()) { + throw new TimeoutException(String.format("Unable to abort %s in timely manner.",stateBeingAbborted)); + } } /** - * Schedule a service for async task with the passed in parameters - * @param rpc of the REST API call, decides how to schedule the service + * Schedule a {@link BaseActionRunnable} to begin async execution. This is the Parent {@link Runnable} for the + * children that are submitted by {@link #submitBaseSubCallable(Callable)} + * + * The currently executing {@link BaseActionRunnable} must fully be terminated before the next can be scheduled. + * This means all Tasks' {@link MyFuture#isDone()} must equal true and all threads must return to their respective + * thread pools. + * * @param runnable of the to be scheduled service. - * @return the reference of the scheduled task + * @param cancelCallBack to be invoked when + * {@link #cancelBaseActionRunnable(AppcOam.RPC, AppcOamStates, long, TimeUnit)} is invoked. + * @param initialDelayMillis the time to delay first execution + * @param delayMillis the delay between the termination of one + * execution and the commencement of the next + * @return The {@link BaseActionRunnable}'s {@link Future} + * @throws IllegalStateException if there is currently executing Task */ - public Future<?> scheduleAsyncTask(final AppcOam.RPC rpc, final BaseActionRunnable runnable) { - int initialDelay, interval; - switch (rpc) { - case maintenance_mode: - initialDelay = interval =MMODE_TASK_DELAY; - break; - case start: - case stop: - case restart: - initialDelay = COMMON_INITIAL_DELAY; - interval = COMMON_INTERVAL; - break; - default: - // should not get here. Log it and return null - logDebug(String.format("Cannot scheudle task for unsupported RPC(%s).", rpc.name())); - return null; - } - - // Always cancel existing async task - if (backgroundOamTask != null) { - logDebug("Cancelling background task in schedule task."); - backgroundOamTask.cancel(true); - if (taskRunnable != null) { - taskRunnable.abortRunnable(rpc); - } + public synchronized Future<?> scheduleBaseRunnable(final Runnable runnable, + final Consumer<AppcOam.RPC> cancelCallBack, + long initialDelayMillis, + long delayMillis) + throws IllegalStateException { + + if (backgroundBaseRunnableFuture != null && !backgroundBaseRunnableFuture.isDone()) { + throw new IllegalStateException("Unable to schedule background task when one is already running. All task must fully terminated before another can be scheduled. "); } - taskRunnable = runnable; - backgroundOamTask = scheduledExecutorService.scheduleWithFixedDelay( - runnable, initialDelay, interval, TimeUnit.MILLISECONDS); + this.cancelCallBackForBaseRunnable = cancelCallBack; - return backgroundOamTask; - } + backgroundBaseRunnableFuture = new MyFuture(runnable) { + /** + * augments the cancel operation to cancel all subTack too, + */ + @Override + public boolean cancel(final boolean mayInterruptIfRunning) { + boolean cancel; + synchronized (AsyncTaskHelper.this) { + cancel = super.cancel(mayInterruptIfRunning); + myFutureSet.stream().filter(f->!this.equals(f)).forEach(f->f.cancel(mayInterruptIfRunning)); + } + return cancel; + } - Future<?> submitBundleLcOperation(final Callable callable) { - return bundleOperationService.submit(callable); + /** + * augments the isDone operation to return false until all subTask have completed too. + */ + @Override + public boolean isDone() { + synchronized (AsyncTaskHelper.this) { + return myFutureSet.isEmpty(); + } + } + }; + backgroundBaseRunnableFuture.setFuture( + scheduledExecutorService.scheduleWithFixedDelay( + backgroundBaseRunnableFuture, initialDelayMillis, delayMillis, TimeUnit.MILLISECONDS) + ); + return backgroundBaseRunnableFuture; } /** - * Cancle a previously schedule task. If the task is the same as backgroundOamTask, set it to null. - * @param task to be canceled + * Submits children {@link Callable} to be executed as soon as possible, A parent must have been scheduled + * previously via {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)} + * @param callable the Callable to be submitted + * @return The {@link Callable}'s {@link Future} */ - public void cancelAsyncTask(Future<?> task) { - task.cancel(false); - if (task == backgroundOamTask) { - backgroundOamTask = null; - taskRunnable = null; - logDebug("Cancelling background task in cancel task."); + synchronized Future<?> submitBaseSubCallable(final Callable callable) { + + if (backgroundBaseRunnableFuture == null + || backgroundBaseRunnableFuture.isCancelled() + || backgroundBaseRunnableFuture.isDone()){ + throw new IllegalStateException("Unable to schedule subCallable when a base Runnable is not running."); + } + + //Make sure the pool is ready to go + if(bundleOperationService.getPoolSize() != bundleOperationService.getMaximumPoolSize()){ + bundleOperationService.setCorePoolSize(bundleOperationService.getMaximumPoolSize()); + bundleOperationService.prestartAllCoreThreads(); + bundleOperationService.setCorePoolSize(0); } + + MyFuture<?> myFuture = new MyFuture(callable); + myFuture.setFuture(bundleOperationService.submit((Callable)myFuture)); + return myFuture; } /** @@ -173,4 +259,132 @@ public class AsyncTaskHelper { logger.debug(String.format(message, args)); } } + + /** + * This class has two purposes. First it insures {@link #isDone()} only returns true if the deligate is not + * currently running and will not be running in the future: See Java BUG JDK-8073704 Second this class maintains + * the {@link #myFutureSet } by insurring that itself is removed when {@link #isDone()} returns true. + * + * See {@link #scheduleBaseRunnable(Runnable, Consumer, long, long)} and {@link #submitBaseSubCallable(Callable)} + * for usage of this class + */ + private class MyFuture<T> implements Future<T>, Runnable, Callable<T> { + + private Future<T> future; + private final Runnable runnable; + private final Callable<T> callable; + private boolean isRunning; + + MyFuture(Runnable runnable) { + this.runnable = runnable; + this.callable = null; + myFutureSet.add(this); + } + + MyFuture(Callable<T> callable) { + this.runnable = null; + this.callable = callable; + myFutureSet.add(this); + } + + void setFuture(Future<T> future) { + this.future = future; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + synchronized (AsyncTaskHelper.this) { + if (!isRunning) { + myFutureSetRemove(); + } + + return future.cancel(mayInterruptIfRunning); + } + } + + @Override + public boolean isCancelled() { + synchronized (AsyncTaskHelper.this) { + return future.isCancelled(); + } + } + + @Override + public boolean isDone() { + synchronized (AsyncTaskHelper.this) { + return future.isDone() && !isRunning; + } + } + + @Override + public T get() throws InterruptedException, ExecutionException { + return future.get(); + } + + @Override + public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return future.get(timeout, unit); + } + + @Override + public void run() { + synchronized (AsyncTaskHelper.this) { + if(future.isCancelled()){ + return; + } + isRunning = true; + } + try { + runnable.run(); + } finally { + synchronized (AsyncTaskHelper.this) { + isRunning = false; + + //The Base Runnable is expected to run again. + //unless it has been canceled. + //so only removed if it is canceled. + if (future.isCancelled()) { + myFutureSetRemove(); + } + } + } + } + + @Override + public T call() throws Exception { + synchronized (AsyncTaskHelper.this) { + if(future.isCancelled()){ + throw new CancellationException(); + } + isRunning = true; + } + try { + return callable.call(); + } finally { + synchronized (AsyncTaskHelper.this){ + isRunning = false; + myFutureSetRemove(); + } + } + } + + + /** + * Removes this from the the myFutureSet. + * When all the BaseActionRunnable is Done notify any thread waiting in + * {@link AsyncTaskHelper#cancelBaseActionRunnable(AppcOam.RPC, AppcOamStates, long, TimeUnit)} + */ + void myFutureSetRemove(){ + synchronized (AsyncTaskHelper.this) { + myFutureSet.remove(this); + if(myFutureSet.isEmpty()){ + backgroundBaseRunnableFuture = null; + cancelCallBackForBaseRunnable = null; + AsyncTaskHelper.this.notifyAll(); + + } + } + } + + } } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/BundleHelper.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/BundleHelper.java index 7fbb3c453..74159bd65 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/BundleHelper.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/BundleHelper.java @@ -28,6 +28,7 @@ import com.att.eelf.configuration.EELFLogger; import org.apache.commons.lang3.ArrayUtils; import org.openecomp.appc.exceptions.APPCException; import org.openecomp.appc.oam.AppcOam; +import org.openecomp.appc.oam.processor.BaseCommon; import org.openecomp.appc.statemachine.impl.readers.AppcOamStates; import org.osgi.framework.Bundle; import org.osgi.framework.BundleContext; @@ -73,8 +74,9 @@ public class BundleHelper { */ public boolean bundleOperations(AppcOam.RPC rpc, Map<String, Future<?>> threads, - AsyncTaskHelper taskHelper) - throws APPCException { + AsyncTaskHelper taskHelper, + BaseCommon baseCommon) + throws APPCException { long mStartTime = System.currentTimeMillis(); logDebug(String.format("Entering OAM bundleOperations with rpc (%s).", rpc.name())); @@ -88,7 +90,6 @@ public class BundleHelper { boolean isBundleOperationComplete = true; Map<String, Bundle> appcLcmBundles = getAppcLcmBundles(); - taskHelper.addThreadsToPool(); for (Map.Entry<String, Bundle> bundleEntry : appcLcmBundles.entrySet()) { String bundleName = bundleEntry.getKey(); Bundle bundle = bundleEntry.getValue(); @@ -99,30 +100,29 @@ public class BundleHelper { // such as when a Stop request is receive while APPC is still trying to Start Up. if (!stateHelper.isSameState(originalState)) { logger.warn("OAM %s bundle operation aborted since OAM state is no longer %s!", - originalState.name()); + originalState.name()); isBundleOperationComplete = false; break; } } threads.put(bundleName, - taskHelper.submitBundleLcOperation(new BundleTask(rpc, bundle))); + taskHelper.submitBaseSubCallable(new BundleTask(rpc, bundle,baseCommon))); } - taskHelper.removeThreadsFromPoolWhenDone(); logDebug(String.format("Leaving OAM bundleOperations with rpc (%s) with complete(%s), elasped (%d) ms.", - rpc.name(), Boolean.toString(isBundleOperationComplete), getElaspeTimeMs(mStartTime))); + rpc.name(), Boolean.toString(isBundleOperationComplete), getElapseTimeMs(mStartTime))); return isBundleOperationComplete; } - private long getElaspeTimeMs(long mStartTime) { + private long getElapseTimeMs(long mStartTime) { return System.currentTimeMillis() - mStartTime; } /** * Check if all BundleTasks are completed - * @param bundleNameFutureMap with bundler name and BundleTask Future object + * @param bundleNameFutureMap with bundle name and BundleTask Future object * @return true if all are done, otherwise, false */ public boolean isAllTaskDone(Map<String, Future<?>> bundleNameFutureMap) { @@ -131,20 +131,22 @@ public class BundleHelper { } /** - * Cancel BunldeTasks which are not finished - * @param bundleNameFutureMap with bundler name and BundleTask Future object + * Cancel BundleTasks which are not finished + * @param bundleNameFutureMap with bundle name and BundleTask Future object */ public void cancelUnfinished(Map<String, Future<?>> bundleNameFutureMap) { - bundleNameFutureMap.values().stream().filter((f) -> !f.isDone()).forEach((f) -> f.cancel(true)); + bundleNameFutureMap.values().stream().filter((f) + -> !f.isDone()).forEach((f) + -> f.cancel(true)); } /** * Get number of failed BundleTasks - * @param bundleNameFurtureMap with bundler name and BundleTask Future object + * @param bundleNameFutureMap with bundle name and BundleTask Future object * @return number(long) of the failed BundleTasks */ - public long getFailedMetrics(Map<String, Future<?>> bundleNameFurtureMap) { - return bundleNameFurtureMap.values().stream().map((f) -> { + public long getFailedMetrics(Map<String, Future<?>> bundleNameFutureMap) { + return bundleNameFutureMap.values().stream().map((f) -> { try { return f.get(); } catch (Exception e) { @@ -168,10 +170,10 @@ public class BundleHelper { BundleFilter bundleList = new BundleFilter(bundlesToStop, regExBundleNotStop, getBundleList()); logger.info(String.format("(%d) APPC bundles to Stop/Start: %s.", bundleList.getBundlesToStop().size(), - bundleList.getBundlesToStop().toString())); + bundleList.getBundlesToStop().toString())); logger.debug(String.format("(%d) APPC bundles that won't be Stopped/Started: %s.", - bundleList.getBundlesToNotStop().size(), bundleList.getBundlesToNotStop().toString())); + bundleList.getBundlesToNotStop().size(), bundleList.getBundlesToNotStop().toString())); return bundleList.getBundlesToStop(); } @@ -229,17 +231,21 @@ public class BundleHelper { private Bundle bundle; private String bundleName; private String actionName; + private final BaseCommon baseCommon; - BundleTask(AppcOam.RPC rpcIn, Bundle bundleIn) { + BundleTask(AppcOam.RPC rpcIn, Bundle bundleIn, BaseCommon baseCommon) { rpc = rpcIn; actionName = rpc.getAppcOperation().toString(); bundle = bundleIn; bundleName = bundle.getSymbolicName(); + this.baseCommon = baseCommon; } @Override public BundleTask call() throws Exception { try { + baseCommon.setInitialLogProperties(); + long bundleOperStartTime = System.currentTimeMillis(); logDebug(String.format("OAM %s bundle %s ===>", actionName, bundleName)); switch (rpc) { @@ -253,12 +259,15 @@ public class BundleHelper { // should do nothing } logDebug(String.format("OAM %s bundle %s completed <=== elasped %d", - actionName, bundleName, getElaspeTimeMs(bundleOperStartTime))); + actionName, bundleName, getElapseTimeMs(bundleOperStartTime))); } catch (BundleException e) { logger.error(String.format("Exception encountered when OAM %s bundle %s ", - actionName, bundleName), e); + actionName, bundleName), e); failException = e; } + finally { + baseCommon.clearRequestLogProperties(); + } return this; } } diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/ConfigurationHelper.java b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/ConfigurationHelper.java index c465b9b10..6e6ab036b 100644 --- a/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/ConfigurationHelper.java +++ b/appc-oam/appc-oam-bundle/src/main/java/org/openecomp/appc/oam/util/ConfigurationHelper.java @@ -30,12 +30,17 @@ import org.openecomp.appc.Constants; import org.openecomp.appc.configuration.Configuration; import org.openecomp.appc.configuration.ConfigurationFactory; +import java.util.concurrent.TimeUnit; + /** * Utility class provides general configuration helps */ public class ConfigurationHelper { final static String PROP_KEY_APPC_NAME = Constants.PROPERTY_APPLICATION_NAME; final static String PROP_KEY_METRIC_STATE = "metric.enabled"; + private final String OAM_OPERATION_TIMEOUT_SECOND = "appc.OAM.api.timeout"; + /** Default operation timeout set to 1 minute */ + private final int DEFAULT_OAM_OPERATION_TIMEOUT = 60; private final EELFLogger logger; private Configuration configuration = ConfigurationFactory.getConfiguration(); @@ -57,7 +62,7 @@ public class ConfigurationHelper { } /** - * Read property value of a specified proeprty key + * Read property value of a specified property key * * @param propertyKey string of the property key * @return String[] of the property values associated with the propertyKey @@ -77,4 +82,23 @@ public class ConfigurationHelper { } return new String[]{propertyValue}; } + + + + + + /** + * This method returns timeout in milliseconds. The source is chosen in the following order: + * The overrideTimeoutSeconds argument + * or {@link #OAM_OPERATION_TIMEOUT_SECOND} found in the configuration file + * or the {@link #DEFAULT_OAM_OPERATION_TIMEOUT} + * @param overrideTimeoutSeconds or null to us the other sources + * @return timeout in milliseconds + */ + public long getOAMOperationTimeoutValue(Integer overrideTimeoutSeconds) { + return overrideTimeoutSeconds == null ? + getConfig().getIntegerProperty(OAM_OPERATION_TIMEOUT_SECOND, DEFAULT_OAM_OPERATION_TIMEOUT) * 1000 + : + TimeUnit.MILLISECONDS.toMillis(overrideTimeoutSeconds); + } } |