| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you 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. |
| */ |
| package org.apache.brooklyn.core.workflow; |
| |
| import com.fasterxml.jackson.annotation.JsonIgnore; |
| import com.fasterxml.jackson.annotation.JsonInclude; |
| import com.fasterxml.jackson.annotation.JsonProperty; |
| import com.fasterxml.jackson.databind.annotation.JsonDeserialize; |
| import com.google.common.collect.Iterables; |
| import com.google.common.reflect.TypeToken; |
| import org.apache.brooklyn.api.entity.Entity; |
| import org.apache.brooklyn.api.mgmt.ManagementContext; |
| import org.apache.brooklyn.api.mgmt.Task; |
| import org.apache.brooklyn.api.mgmt.classloading.BrooklynClassLoadingContext; |
| import org.apache.brooklyn.api.objs.BrooklynObject; |
| import org.apache.brooklyn.api.sensor.AttributeSensor; |
| import org.apache.brooklyn.config.ConfigKey; |
| import org.apache.brooklyn.core.effector.Effectors; |
| import org.apache.brooklyn.core.entity.Entities; |
| import org.apache.brooklyn.core.entity.EntityAdjuncts; |
| import org.apache.brooklyn.core.entity.EntityInternal; |
| import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; |
| import org.apache.brooklyn.core.resolve.jackson.JsonPassThroughDeserializer; |
| import org.apache.brooklyn.core.sensor.Sensors; |
| import org.apache.brooklyn.core.typereg.RegisteredTypes; |
| import org.apache.brooklyn.core.workflow.store.WorkflowRetentionAndExpiration; |
| import org.apache.brooklyn.core.workflow.store.WorkflowStatePersistenceViaSensors; |
| import org.apache.brooklyn.core.workflow.utils.WorkflowRetentionParser; |
| import org.apache.brooklyn.util.collections.MutableList; |
| import org.apache.brooklyn.util.collections.MutableMap; |
| import org.apache.brooklyn.util.collections.MutableSet; |
| import org.apache.brooklyn.util.core.config.ConfigBag; |
| import org.apache.brooklyn.util.core.flags.BrooklynTypeNameResolution; |
| import org.apache.brooklyn.util.core.predicates.DslPredicates; |
| import org.apache.brooklyn.util.core.task.DynamicTasks; |
| import org.apache.brooklyn.util.core.task.TaskBuilder; |
| import org.apache.brooklyn.util.core.task.TaskTags; |
| import org.apache.brooklyn.util.core.task.Tasks; |
| import org.apache.brooklyn.util.core.text.TemplateProcessor; |
| import org.apache.brooklyn.util.exceptions.Exceptions; |
| import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException; |
| import org.apache.brooklyn.util.guava.Maybe; |
| import org.apache.brooklyn.util.javalang.Threads; |
| import org.apache.brooklyn.util.text.Strings; |
| import org.apache.brooklyn.util.time.Duration; |
| import org.apache.brooklyn.util.time.Time; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.annotation.Nullable; |
| import java.time.Instant; |
| import java.util.*; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.BiConsumer; |
| import java.util.function.Consumer; |
| import java.util.function.Function; |
| import java.util.function.Supplier; |
| |
| import static org.apache.brooklyn.core.workflow.WorkflowReplayUtils.ReplayResumeDepthCheck.RESUMABLE_WHENEVER_NESTED_WORKFLOWS_PRESENT; |
| |
| @JsonInclude(JsonInclude.Include.NON_NULL) |
| public class WorkflowExecutionContext { |
| |
| private static final Logger log = LoggerFactory.getLogger(WorkflowExecutionContext.class); |
| |
| public static final String LABEL_FOR_ERROR_HANDLER = "error-handler"; |
| |
| public static final int STEP_INDEX_FOR_START = -1; |
| public static final int STEP_INDEX_FOR_END = -2; |
| public static final int STEP_INDEX_FOR_ERROR_HANDLER = -3; |
| |
| public static final String STEP_TARGET_NAME_FOR_START = "start"; |
| public static final String STEP_TARGET_NAME_FOR_END = "end"; |
| public static final String STEP_TARGET_NAME_FOR_LAST = "last"; |
| public static final String STEP_TARGET_NAME_FOR_HERE = "here"; |
| public static final String STEP_TARGET_NAME_FOR_EXIT = "exit"; |
| public static final String STEP_TARGET_NAME_FOR_DEFAULT = "default"; |
| |
| public static final Map<String, Function<WorkflowExecutionContext,Integer>> PREDEFINED_NEXT_TARGETS = MutableMap.<String, Function<WorkflowExecutionContext,Integer>>of( |
| STEP_TARGET_NAME_FOR_START, c -> c==null? STEP_INDEX_FOR_START : 0, |
| STEP_TARGET_NAME_FOR_END, c -> c==null ? STEP_INDEX_FOR_END : c.stepsDefinition.size(), |
| STEP_TARGET_NAME_FOR_LAST, c -> c==null ? null : c.replayableLastStep, |
| STEP_TARGET_NAME_FOR_HERE, c -> c==null ? null : null, |
| STEP_TARGET_NAME_FOR_EXIT, c -> c==null ? null : null, |
| STEP_TARGET_NAME_FOR_DEFAULT, c -> c==null ? null : c.currentStepIndex+1).asUnmodifiable(); |
| |
| String name; |
| @Nullable BrooklynObject adjunct; |
| Entity entity; |
| |
| public enum WorkflowStatus { |
| STAGED(false, false, false, false), |
| RUNNING(true, false, false, false), |
| SUCCESS(true, true, false, true), |
| /** useful information, usually cannot persisted by the time we've set this the first time, but could set on rebind */ ERROR_SHUTDOWN(true, true, true, false), |
| /** task failed because entity destroyed */ ERROR_ENTITY_DESTROYED(true, true, true, true), |
| /** task cancelled, timeout, or other interrupt, usually recursively (but not entity destroyed or server shutdown) */ ERROR_CANCELLED(true, true, true, true), |
| /** any other error, e.g. workflow step failed or data not immediately available (the interrupt used internally is not relevant) */ ERROR(true, true, true, true); |
| |
| public final boolean started; |
| public final boolean ended; |
| public final boolean error; |
| public final boolean expirable; |
| |
| WorkflowStatus(boolean started, boolean ended, boolean error, boolean expirable) { this.started = started; this.ended = ended; this.error = error; this.expirable = expirable; } |
| } |
| |
| WorkflowStatus status; |
| Instant lastStatusChangeTime; |
| |
| @JsonIgnore private transient WorkflowExecutionContext parent; |
| private BrooklynTaskTags.WorkflowTaskTag parentTag; |
| |
| // should be treated as raw json |
| @JsonDeserialize(contentUsing = JsonPassThroughDeserializer.class) |
| List<Object> stepsDefinition; |
| |
| DslPredicates.DslPredicate condition; |
| |
| @JsonInclude(JsonInclude.Include.NON_EMPTY) |
| Map<String,Object> input = MutableMap.of(); |
| @JsonIgnore // persist as sensor but not via REST in case it has secrets resolved |
| Map<String,Object> inputResolved = MutableMap.of(); |
| |
| Object outputDefinition; |
| Object output; |
| |
| Object lock; |
| |
| Duration timeout; |
| Object onError; |
| |
| String workflowId; |
| /** current or most recent executing task created for this workflow, corresponding to task */ |
| String taskId; |
| transient Task<Object> task; |
| |
| @JsonProperty("retention") |
| WorkflowRetentionAndExpiration.WorkflowRetentionSettings retention; |
| |
| /** all tasks created for this workflow */ |
| Set<WorkflowReplayUtils.WorkflowReplayRecord> replays = MutableSet.of(); |
| transient WorkflowReplayUtils.WorkflowReplayRecord replayCurrent = null; |
| /** null if no replay point, otherwise step number of the last completed replay point, or -1 if should replay from start and -2 if completed successfully */ |
| Integer replayableLastStep; |
| Boolean replayableFromStart; |
| Boolean replayableAutomatically; |
| Boolean replayableDisabled; |
| Boolean idempotentAll; |
| |
| Integer currentStepIndex; |
| Integer previousStepIndex; |
| String previousStepTaskId; |
| |
| WorkflowStepInstanceExecutionContext currentStepInstance; |
| |
| /** set if an error handler is the last thing which ran */ |
| String errorHandlerTaskId; |
| /** set for the last _step_ inside the error handler */ |
| WorkflowStepInstanceExecutionContext errorHandlerContext; |
| |
| Map<Integer, OldStepRecord> oldStepInfo = MutableMap.of(); |
| |
| @JsonInclude(JsonInclude.Include.NON_EMPTY) |
| public static class OldStepRecord { |
| /** count of runs started */ |
| int countStarted = 0; |
| /** count of runs completed */ |
| int countCompleted = 0; |
| |
| /** context for last _completed_ instance of step */ |
| WorkflowStepInstanceExecutionContext context; |
| /** is step replayable */ |
| Boolean replayableFromHere; |
| /** scratch for last _started_ instance of step */ |
| Map<String,Object> workflowScratch; |
| /** steps that immediately preceded this, updated when _this_ step started, with most recent first */ |
| Set<Integer> previous; |
| /** steps that immediately followed this, updated when _next_ step started, with most recent first */ |
| Set<Integer> next; |
| |
| String previousTaskId; |
| String nextTaskId; |
| } |
| |
| Map<String,Object> workflowScratchVariables = MutableMap.of(); |
| |
| @JsonInclude(JsonInclude.Include.NON_EMPTY) |
| Map<String,List<Instant>> retryRecords = MutableMap.of(); |
| |
| // deserialization constructor |
| private WorkflowExecutionContext() {} |
| |
| public static WorkflowExecutionContext newInstancePersisted(BrooklynObject entityOrAdjunctWhereRunning, WorkflowContextType wcType, String name, ConfigBag paramsDefiningWorkflow, |
| Collection<ConfigKey<?>> extraConfigKeys, ConfigBag extraInputs, Map<String, Object> optionalTaskFlags) { |
| WorkflowExecutionContext w = newInstanceUnpersistedWithParent(entityOrAdjunctWhereRunning, null, wcType, name, paramsDefiningWorkflow, extraConfigKeys, extraInputs, optionalTaskFlags); |
| w.persist(); |
| return w; |
| } |
| |
| public enum WorkflowContextType { |
| SENSOR, EFFECTOR, POLICY, NESTED_WORKFLOW, OTHER |
| } |
| |
| public static WorkflowExecutionContext newInstanceUnpersistedWithParent(BrooklynObject entityOrAdjunctWhereRunning, WorkflowExecutionContext parent, |
| WorkflowContextType wcType, String name, ConfigBag paramsDefiningWorkflow, |
| Collection<ConfigKey<?>> extraConfigKeys, ConfigBag extraInputs, Map<String, Object> optionalTaskFlags) { |
| |
| // parameter defs |
| Map<String,ConfigKey<?>> parameters = MutableMap.of(); |
| Maybe<BrooklynClassLoadingContext> loader = RegisteredTypes.getClassLoadingContextMaybe(entityOrAdjunctWhereRunning); |
| Effectors.parseParameters(paramsDefiningWorkflow.get(WorkflowCommonConfig.PARAMETER_DEFS), loader.orNull()).forEach(p -> parameters.put(p.getName(), Effectors.asConfigKey(p))); |
| if (extraConfigKeys!=null) extraConfigKeys.forEach(p -> parameters.put(p.getName(), p)); |
| |
| // inputs, unresolved first |
| ConfigBag inputRaw = ConfigBag.newInstance(); |
| inputRaw.putAll(paramsDefiningWorkflow.get(WorkflowCommonConfig.INPUT)); |
| if (extraInputs!=null) inputRaw.putAll(extraInputs.getAllConfig()); |
| parameters.values().forEach(p -> { |
| if (p.hasDefaultValue() && !inputRaw.containsKey(p.getName())) inputRaw.put((ConfigKey)p, p.getDefaultValue()); |
| }); |
| |
| MutableMap<String,Object> input = MutableMap.of(); |
| inputRaw.forEach( (k,v) -> { |
| ConfigKey<?> kc = parameters.get(k); |
| // coerce, but don't freemarker resolve inputs because that's the job of the caller (e.g. in nested workflow, inputs resolved relative to calling workflow) |
| Object v2 = kc == null ? v : inputRaw.get(kc); |
| input.put(k, v2); |
| }); |
| |
| WorkflowExecutionContext w = new WorkflowExecutionContext(entityOrAdjunctWhereRunning, parent, name, |
| paramsDefiningWorkflow.get(WorkflowCommonConfig.STEPS), |
| input, |
| paramsDefiningWorkflow.get(WorkflowCommonConfig.OUTPUT), |
| WorkflowReplayUtils.updaterForReplayableAtWorkflow(paramsDefiningWorkflow, wcType == WorkflowContextType.NESTED_WORKFLOW), |
| optionalTaskFlags); |
| |
| w.getStepsResolved(); // ensure steps resolve at this point; should be true even if condition doesn't apply (though input might not be valid without condition) |
| |
| w.retention = WorkflowRetentionParser.parse(paramsDefiningWorkflow.get(WorkflowCommonConfig.RETENTION), w).init(w); |
| w.lock = paramsDefiningWorkflow.get(WorkflowCommonConfig.LOCK); |
| w.timeout = paramsDefiningWorkflow.get(WorkflowCommonConfig.TIMEOUT); |
| w.onError = paramsDefiningWorkflow.get(WorkflowCommonConfig.ON_ERROR); |
| // fail fast if error steps not resolveable |
| WorkflowStepResolution.resolveSubSteps(w.getManagementContext(), "error handling", WorkflowErrorHandling.wrappedInListIfNecessaryOrNullIfEmpty(w.onError)); |
| |
| // some fields need to be resolved at setting time, in the context of the workflow |
| w.setCondition(w.resolveWrapped(WorkflowExpressionResolution.WorkflowExpressionStage.WORKFLOW_STARTING_POST_INPUT, paramsDefiningWorkflow.getStringKey(WorkflowCommonConfig.CONDITION.getName()), WorkflowCommonConfig.CONDITION.getTypeToken())); |
| |
| // finished -- checkpoint noting this has been created but not yet started |
| w.updateStatus(WorkflowStatus.STAGED); |
| return w; |
| } |
| |
| protected WorkflowExecutionContext(BrooklynObject entityOrAdjunctWhereRunning, WorkflowExecutionContext parent, String name, |
| List<Object> stepsDefinition, Map<String,Object> input, Object output, |
| Consumer<WorkflowExecutionContext> replayableInitializer, Map<String, Object> optionalTaskFlags) { |
| initParent(parent); |
| this.name = name; |
| this.adjunct = entityOrAdjunctWhereRunning instanceof Entity ? null : entityOrAdjunctWhereRunning; |
| this.entity = entityOrAdjunctWhereRunning instanceof Entity ? (Entity)entityOrAdjunctWhereRunning : ((EntityAdjuncts.EntityAdjunctProxyable)entityOrAdjunctWhereRunning).getEntity(); |
| this.stepsDefinition = stepsDefinition; |
| |
| this.input = input; |
| this.outputDefinition = output; |
| if (replayableInitializer!=null) replayableInitializer.accept(this); |
| |
| TaskBuilder<Object> tb = Tasks.builder().dynamic(true); |
| if (optionalTaskFlags!=null) tb.flags(optionalTaskFlags); |
| if (Strings.isBlank(tb.getDisplayName())) tb.displayName(name); |
| task = tb.body(new Body()).build(); |
| WorkflowReplayUtils.updateOnWorkflowStartOrReplay(this, task, "initial run", null); |
| workflowId = taskId = task.getId(); |
| TaskTags.addTagDynamically(task, BrooklynTaskTags.WORKFLOW_TAG); |
| TaskTags.addTagDynamically(task, BrooklynTaskTags.tagForWorkflow(this)); |
| |
| // currently workflow ID is the same as the task ID assigned initially. (but if replayed they will be different.) |
| // there is no deep reason or need for this, it is just convenient, and used for tests. |
| //this.workflowId = Identifiers.makeRandomId(8); |
| } |
| |
| public void initParent(WorkflowExecutionContext parent) { |
| this.parent = parent; |
| this.parentTag = parent==null ? null : BrooklynTaskTags.tagForWorkflow(parent); |
| } |
| |
| @JsonIgnore |
| public WorkflowExecutionContext getParent() { |
| if (parent==null && parentTag!=null) { |
| Entity entity = getManagementContext().getEntityManager().getEntity(parentTag.getEntityId()); |
| if (entity==null) { |
| log.warn("Parent workflow "+parentTag+" for "+this+" is on an entity no longer known; unparenting this workflow"); |
| parentTag = null; |
| } else { |
| parent = new WorkflowStatePersistenceViaSensors(getManagementContext()).getWorkflows(entity).get(parentTag.getWorkflowId()); |
| if (parent==null) { |
| log.warn("Parent workflow "+parentTag+" for "+this+" is no longer known; unparenting this workflow"); |
| parentTag = null; |
| } |
| } |
| } |
| return parent; |
| } |
| |
| public static void validateSteps(ManagementContext mgmt, List<WorkflowStepDefinition> steps, boolean alreadyValidatedIndividualSteps) { |
| if (!alreadyValidatedIndividualSteps) { |
| steps.forEach(w -> w.validateStep(mgmt, null)); |
| } |
| |
| computeStepsWithExplicitIdById(steps); |
| } |
| |
| static Map<String,Pair<Integer,WorkflowStepDefinition>> computeStepsWithExplicitIdById(List<WorkflowStepDefinition> steps) { |
| Map<String,Pair<Integer,WorkflowStepDefinition>> stepsWithExplicitId = MutableMap.of(); |
| for (int i = 0; i<steps.size(); i++) { |
| WorkflowStepDefinition s = steps.get(i); |
| if (s.id != null) { |
| if (PREDEFINED_NEXT_TARGETS.containsKey(s.id.toLowerCase())) |
| throw new IllegalStateException("Token '" + s + "' is a reserved word and cannot be used as a step ID"); |
| |
| Pair<Integer, WorkflowStepDefinition> old = stepsWithExplicitId.put(s.id, Pair.of(i, s)); |
| if (old != null) throw new IllegalStateException("Same step ID '" + s + "' used for multiple steps ("+(old.getLeft()+1)+" and "+(i+1)+")"); |
| } |
| } |
| return stepsWithExplicitId; |
| } |
| |
| public void setCondition(DslPredicates.DslPredicate condition) { |
| this.condition = condition; |
| } |
| |
| @Override |
| public String toString() { |
| return "Workflow<" + name + " - " + workflowId + ">"; |
| } |
| |
| @JsonIgnore |
| public BrooklynObject getEntityOrAdjunctWhereRunning() { |
| if (adjunct!=null) return adjunct; |
| return entity; |
| } |
| |
| public BrooklynTaskTags.WorkflowTaskTag getParentTag() { |
| return parentTag; |
| } |
| |
| public Map<String, Object> getWorkflowScratchVariables() { |
| return workflowScratchVariables; |
| } |
| |
| public Map<String, List<Instant>> getRetryRecords() { |
| return retryRecords; |
| } |
| |
| @JsonIgnore |
| public Object getConditionTarget() { |
| if (getWorkflowScratchVariables()!=null) { |
| Object v = getWorkflowScratchVariables().get("target"); |
| // should we also set the entity? otherwise it will take from the task. but that should only apply |
| // in a task where the context entity is set, so for now rely on that. |
| if (v!=null) return v; |
| } |
| return getEntityOrAdjunctWhereRunning(); |
| } |
| |
| @JsonIgnore |
| public Maybe<Task<Object>> getTask(boolean checkCondition) { |
| if (checkCondition && condition!=null) { |
| if (!condition.apply(getConditionTarget())) return Maybe.absent(new IllegalStateException("This workflow cannot be run at present: condition not satisfied")); |
| } |
| |
| if (task==null) { |
| if (taskId !=null) { |
| task = (Task<Object>) getManagementContext().getExecutionManager().getTask(taskId); |
| } |
| if (task==null) { |
| return Maybe.absent(new IllegalStateException("Task for "+this+" no longer available")); |
| } |
| } |
| return Maybe.of(task); |
| } |
| |
| public Factory factory(boolean allowInternallyEvenIfDisabled) { |
| return new Factory(allowInternallyEvenIfDisabled); |
| } |
| |
| public class Factory { |
| private final boolean allowInternallyEvenIfDisabled; |
| |
| protected Factory(boolean allowInternallyEvenIfDisabled) { |
| this.allowInternallyEvenIfDisabled = allowInternallyEvenIfDisabled; |
| } |
| |
| public WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayingFromStep(int stepIndex0, String reason, boolean forced) { |
| if (!forced) checkNotDisabled(); |
| int stepIndex = stepIndex0; |
| if (!forced) { |
| stepIndex = Maybe.ofDisallowingNull(WorkflowReplayUtils.findNearestReplayPoint(WorkflowExecutionContext.this, stepIndex0)) |
| .orThrow(() -> new IllegalStateException("Workflow is not replayable: no replay points found backtracking from " + stepIndex0)); |
| log.debug("Request to replay from step " + stepIndex0 + ", nearest replay point is " + stepIndex); |
| } |
| return new WorkflowStepDefinition.ReplayContinuationInstructions(stepIndex, reason, null, forced); |
| } |
| |
| public WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayingFromLastReplayable(String reason, boolean forced) { |
| return makeInstructionsForReplayingFromStep(replayableLastStep != null ? replayableLastStep : STEP_INDEX_FOR_START, reason, forced); |
| } |
| |
| public WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayingFromStart(String reason, boolean forced) { |
| return makeInstructionsForReplayingFromStep(STEP_INDEX_FOR_START, reason, forced); |
| } |
| |
| public WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayResuming(String reason, boolean forced) { |
| return makeInstructionsForReplayResuming(reason, forced, null); |
| } |
| |
| public WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayResumingForcedWithCustom(String reason, Runnable code) { |
| return makeInstructionsForReplayResuming(reason, true, code); |
| } |
| |
| protected WorkflowStepDefinition.ReplayContinuationInstructions makeInstructionsForReplayResuming(String reason, boolean forced, Runnable code) { |
| if (!forced) checkNotDisabled(); |
| Integer replayFromStep = null; |
| if (currentStepIndex == null) { |
| // not yet started |
| replayFromStep = STEP_INDEX_FOR_START; |
| } else if (currentStepInstance == null || currentStepInstance.stepIndex != currentStepIndex) { |
| // replaying from a different step, or current step which has either not run or completed but didn't save |
| log.debug("Replaying workflow '" + name + "', cannot replay within step " + currentStepIndex + " because step instance not known; will reinitialize then replay that step"); |
| replayFromStep = currentStepIndex; |
| } |
| |
| if (!forced && replayFromStep == null) { |
| // instructions should be made even if subworkflows might reject them; that's the intention of "resume" without force, vs replay from last |
| if (!WorkflowReplayUtils.isReplayResumable(WorkflowExecutionContext.this, RESUMABLE_WHENEVER_NESTED_WORKFLOWS_PRESENT, allowInternallyEvenIfDisabled)) { |
| if (code != null) { |
| // we could allow this, but we don't need it |
| throw new IllegalArgumentException("Cannot supply code to here without forcing as workflow does not support replay resuming at this point"); |
| } |
| log.debug("Request to replay resuming " + WorkflowExecutionContext.this + " at non-idempotent step; rolling back to " + replayableLastStep); |
| if (replayableLastStep == null) { |
| throw new IllegalArgumentException("Cannot replay resuming as there are no replay points and last step " + currentStepIndex + " is not idempotent"); |
| } |
| return makeInstructionsForReplayingFromStep(replayableLastStep, reason, false); |
| } |
| } |
| |
| return new WorkflowStepDefinition.ReplayContinuationInstructions(replayFromStep, reason, code, forced); |
| } |
| |
| public Task<Object> createTaskReplaying(WorkflowStepDefinition.ReplayContinuationInstructions continuationInstructions) { |
| return createTaskReplaying(null, continuationInstructions); |
| } |
| |
| public Task<Object> createTaskReplaying(Runnable intro, WorkflowStepDefinition.ReplayContinuationInstructions continuationInstructions) { |
| if (continuationInstructions==null || !continuationInstructions.forced) checkNotDisabled(); |
| if (task != null && !task.isDone()) { |
| if (!task.isSubmitted()) { |
| if (parent!=null && parent.getReplays().size()>1) { |
| log.debug("Abandoning sub-workflow task that was never submitted, not unusual as parent seems to be replaying: " + task + " for " + WorkflowExecutionContext.this); |
| } else { |
| log.warn("Abandoning workflow task that was never submitted: " + task + " for " + WorkflowExecutionContext.this); |
| } |
| } else { |
| if (isSubmitterAncestor(Tasks.current(), task)) { |
| // not sure we need this check |
| log.debug("Replaying containing workflow " + WorkflowExecutionContext.this + " in task " + task + " which is an ancestor of " + Tasks.current()); |
| } else { |
| log.warn("Unable to replay workflow " + WorkflowExecutionContext.this + " from " + Tasks.current() + " because workflow task " + task + " is ongoing; will delay up to 1s then retry"); |
| // there can be a slight race between tasks ending and the workflow reporting a failure and replaying; |
| // esp in tests, but also in real world, forgive such a situation by delaying the replay for a short time |
| if (!task.blockUntilEnded(Duration.ONE_SECOND)) { |
| log.warn("Unable to replay workflow " + WorkflowExecutionContext.this + " from " + Tasks.current() + " because workflow task " + task + " is ongoing (waited 1s, still ongoing; so rethrowing)"); |
| throw new IllegalStateException("Cannot replay ongoing workflow, given " + continuationInstructions); |
| } |
| } |
| } |
| } |
| |
| String explanation = continuationInstructions.customBehaviorExplanation != null ? continuationInstructions.customBehaviorExplanation : "no explanation"; |
| task = Tasks.builder().dynamic(true).displayName(name + " (" + explanation + ")") |
| .tag(BrooklynTaskTags.tagForWorkflow(WorkflowExecutionContext.this)) |
| .tag(BrooklynTaskTags.WORKFLOW_TAG) |
| .body(new Body(continuationInstructions).withIntro(intro)).build(); |
| WorkflowReplayUtils.updateOnWorkflowStartOrReplay(WorkflowExecutionContext.this, task, continuationInstructions.customBehaviorExplanation, continuationInstructions.stepToReplayFrom); |
| |
| taskId = task.getId(); |
| |
| return task; |
| } |
| |
| public boolean isDisabled() { |
| if (allowInternallyEvenIfDisabled) return false; |
| if (Boolean.TRUE.equals(replayableDisabled)) return true; |
| return false; |
| } |
| |
| public void checkNotDisabled() { |
| if (isDisabled()) throw new IllegalStateException("Replays disabled on "+WorkflowExecutionContext.this); |
| } |
| } |
| |
| private boolean isSubmitterAncestor(Task current, Task<Object> possibleAncestor) { |
| if (current==null) return false; |
| if (current.equals(possibleAncestor)) return true; |
| return isSubmitterAncestor(current.getSubmittedByTask(), possibleAncestor); |
| } |
| |
| public Entity getEntity() { |
| return entity; |
| } |
| |
| @JsonIgnore |
| public ManagementContext getManagementContext() { |
| return ((EntityInternal)getEntity()).getManagementContext(); |
| } |
| |
| @JsonIgnore |
| protected WorkflowStatePersistenceViaSensors getPersister() { |
| return new WorkflowStatePersistenceViaSensors(getManagementContext()); |
| } |
| |
| public void persist() { |
| if (isInErrorHandlerSubWorkflow()) { |
| // currently don't persist error handler sub-workflows |
| return; |
| } |
| getPersister().checkpoint(this); |
| } |
| |
| /** Get the value of the input. Supports Brooklyn DSL resolution but NOT Freemarker resolution. */ |
| public Object getInput(String key) { |
| return getInputMaybe(key, TypeToken.of(Object.class), Maybe.ofAllowingNull(null)).get(); |
| } |
| public <T> Maybe<T> getInputMaybe(String key, TypeToken<T> type, Maybe<T> valueIfUndefined) { |
| if (!input.containsKey(key)) return valueIfUndefined; |
| |
| if (inputResolved.containsKey(key)) return Maybe.ofAllowingNull((T)inputResolved.get(key)); |
| |
| Object v = input.get(key); |
| // normally do DSL resolution/coercion only, not workflow syntax here (as no workflow scope); |
| // except if we are in a nested workflow, we allow resolving from the parent. |
| // (alternatively we could resolve when starting the custom workflow; that might be better.) |
| Maybe<T> vm = null; |
| if (v instanceof String && parent!=null && parent.getCurrentStepInstance()!=null) { |
| try { |
| vm = Maybe.of(parent.getCurrentStepInstance().resolve(WorkflowExpressionResolution.WorkflowExpressionStage.STEP_RUNNING, (String) v, type)); |
| } catch (Exception e) { |
| Exceptions.propagateIfFatal(e); |
| vm = Maybe.absent(e); |
| } |
| } |
| if (vm==null || vm.isAbsent()) { |
| Maybe<T> vm2 = Tasks.resolving(v).as(type).context(getEntity()).immediately(true).deep().getMaybe(); |
| if (vm2.isPresent() || vm==null) vm = vm2; // if errors in both, prefer error in first |
| } |
| if (vm.isPresent()) { |
| if (WorkflowStepInstanceExecutionContext.REMEMBER_RESOLVED_INPUT) { |
| // this will keep spending time resolving, but will resolve the resolved value |
| inputResolved.put(key, vm.get()); |
| } |
| } |
| return vm; |
| } |
| |
| public TypeToken<?> lookupType(String typeName, Supplier<TypeToken<?>> ifUnset) { |
| if (Strings.isBlank(typeName)) return ifUnset.get(); |
| BrooklynClassLoadingContext loader = getEntity() != null ? RegisteredTypes.getClassLoadingContext(getEntity()) : null; |
| return new BrooklynTypeNameResolution.BrooklynTypeNameResolver("workflow", loader, true, true).getTypeToken(typeName); |
| } |
| |
| /** as {@link #resolve(WorkflowExpressionResolution.WorkflowExpressionStage, Object, TypeToken)} but without type coercion */ |
| public Object resolve(WorkflowExpressionResolution.WorkflowExpressionStage stage, String expression) { |
| return resolve(stage, expression, Object.class); |
| } |
| |
| /** as {@link #resolve(WorkflowExpressionResolution.WorkflowExpressionStage, Object, TypeToken)} */ |
| public <T> T resolve(WorkflowExpressionResolution.WorkflowExpressionStage stage, Object expression, Class<T> type) { |
| return resolve(stage, expression, TypeToken.of(type)); |
| } |
| |
| /** resolution of ${interpolation} and $brooklyn:dsl and deferred suppliers, followed by type coercion. |
| * if the type is a string, null is not permitted, otherwise it is. */ |
| public <T> T resolve(WorkflowExpressionResolution.WorkflowExpressionStage stage, Object expression, TypeToken<T> type) { |
| return new WorkflowExpressionResolution(this, stage, false, false).resolveWithTemplates(expression, type); |
| } |
| |
| public <T> T resolveCoercingOnly(WorkflowExpressionResolution.WorkflowExpressionStage stage, Object expression, TypeToken<T> type) { |
| return new WorkflowExpressionResolution(this, stage, false, false).resolveCoercingOnly(expression, type); |
| } |
| |
| /** as {@link #resolve(WorkflowExpressionResolution.WorkflowExpressionStage, Object, TypeToken)}, but returning DSL/supplier for values (so the indication of their dynamic nature is preserved, even if the value returned by it is resolved; |
| * this is needed e.g. for conditions which treat dynamic expressions differently to explicit values) */ |
| public <T> T resolveWrapped(WorkflowExpressionResolution.WorkflowExpressionStage stage, Object expression, TypeToken<T> type) { |
| return new WorkflowExpressionResolution(this, stage, false, true).resolveWithTemplates(expression, type); |
| } |
| |
| /** as {@link #resolve(WorkflowExpressionResolution.WorkflowExpressionStage, Object, TypeToken)}, but waiting on any expressions which aren't ready */ |
| public <T> T resolveWaiting(WorkflowExpressionResolution.WorkflowExpressionStage stage, Object expression, TypeToken<T> type) { |
| return new WorkflowExpressionResolution(this, stage, true, false).resolveWithTemplates(expression, type); |
| } |
| |
| /** resolution of ${interpolation} and $brooklyn:dsl and deferred suppliers, followed by type coercion */ |
| public <T> T resolveConfig(WorkflowExpressionResolution.WorkflowExpressionStage stage, ConfigBag config, ConfigKey<T> key) { |
| Object v = config.getStringKey(key.getName()); |
| if (v==null) return null; |
| return resolve(stage, v, key.getTypeToken()); |
| } |
| |
| public WorkflowStatus getStatus() { |
| return status; |
| } |
| |
| void updateStatus(WorkflowStatus newStatus) { |
| status = newStatus; |
| lastStatusChangeTime = Instant.now(); |
| } |
| |
| @JsonIgnore |
| public Instant getLastStatusChangeTime() { |
| return lastStatusChangeTime; |
| } |
| |
| public Integer getCurrentStepIndex() { |
| return currentStepIndex; |
| } |
| |
| public WorkflowStepInstanceExecutionContext getCurrentStepInstance() { |
| return currentStepInstance; |
| } |
| |
| public Integer getPreviousStepIndex() { |
| return previousStepIndex; |
| } |
| |
| // clear this when error handler is not running |
| transient Object lastErrorHandlerOutput = null; |
| |
| @JsonIgnore |
| public Object getPreviousStepOutput() { |
| if (lastErrorHandlerOutput!=null) return lastErrorHandlerOutput; |
| if (previousStepIndex==null) return null; |
| |
| OldStepRecord last = oldStepInfo.get(previousStepIndex); |
| if (last!=null && last.context!=null) return last.context.output; |
| return null; |
| } |
| |
| public Object getOutput() { |
| return output; |
| } |
| |
| public String getName() { |
| return name; |
| } |
| |
| public String getWorkflowId() { |
| return workflowId; |
| } |
| |
| public String getTaskId() { |
| return taskId; |
| } |
| |
| public Set<WorkflowReplayUtils.WorkflowReplayRecord> getReplays() { |
| return replays; |
| } |
| |
| public Integer getReplayableLastStep() { |
| return replayableLastStep; |
| } |
| |
| public WorkflowStepInstanceExecutionContext getErrorHandlerContext() { |
| return errorHandlerContext; |
| } |
| |
| @JsonIgnore |
| public String getRetentionHash() { |
| if (retention!=null && Strings.isNonBlank(retention.hash)) return retention.hash; |
| if (Strings.isNonBlank(getName())) return getName(); |
| return "anonymous-workflow-"+Math.abs(getStepsDefinition().hashCode()); |
| } |
| |
| public void updateRetentionFrom(WorkflowRetentionAndExpiration.WorkflowRetentionSettings other) { |
| WorkflowRetentionAndExpiration.WorkflowRetentionSettings r = getRetentionSettings(); |
| r.updateFrom(other); |
| retention = r; |
| retentionDefault = null; |
| } |
| |
| @JsonIgnore private transient WorkflowRetentionAndExpiration.WorkflowRetentionSettings retentionDefault; |
| @JsonIgnore |
| public WorkflowRetentionAndExpiration.WorkflowRetentionSettings getRetentionSettings() { |
| if (retention==null) { |
| if (retentionDefault==null) { |
| retentionDefault = new WorkflowRetentionAndExpiration.WorkflowRetentionSettings().init(this); |
| } |
| return retentionDefault; |
| } |
| return retention; |
| } |
| |
| public void markShutdown() { |
| log.debug(this+" was "+this.status+" but now marking as "+WorkflowStatus.ERROR_SHUTDOWN+"; compensating workflow should be triggered shortly"); |
| this.updateStatus(WorkflowStatus.ERROR_SHUTDOWN); |
| // don't persist; that will happen when workflows are kicked off |
| } |
| |
| @JsonIgnore |
| /** Error handlers _could_ launch sub-workflow, but they typically don't */ |
| protected boolean isInErrorHandlerSubWorkflow() { |
| if (getParent()!=null) { |
| if (getParent().getErrorHandlerContext()!=null) { |
| return true; |
| } |
| return getParent().isInErrorHandlerSubWorkflow(); |
| } |
| return false; |
| } |
| |
| /** look in tasks, steps, and replays to find most recent activity */ |
| // keep on jackson serialization for api? |
| public long getMostRecentActivityTime() { |
| AtomicLong result = new AtomicLong(-1); |
| |
| Consumer<Long> consider = l -> { |
| if (l!=null && l>result.get()) result.set(l); |
| }; |
| Consumer<Task> considerTask = task -> { |
| if (task!=null) { |
| consider.accept(task.getEndTimeUtc()); |
| consider.accept(task.getStartTimeUtc()); |
| consider.accept(task.getSubmitTimeUtc()); |
| } |
| }; |
| considerTask.accept(getTask(false).orNull()); |
| |
| Consumer<WorkflowReplayUtils.WorkflowReplayRecord> considerReplay = replay -> { |
| if (replay!=null) { |
| consider.accept(replay.endTimeUtc); |
| consider.accept(replay.startTimeUtc); |
| consider.accept(replay.submitTimeUtc); |
| } |
| }; |
| if (replayCurrent!=null) { |
| considerReplay.accept(replayCurrent); |
| } else if (!replays.isEmpty()) { |
| considerReplay.accept(Iterables.getLast(replays)); |
| } |
| |
| if (currentStepInstance!=null) { |
| considerTask.accept(getManagementContext().getExecutionManager().getTask(currentStepInstance.getTaskId())); |
| } |
| |
| return result.get(); |
| } |
| |
| public List<Object> getStepsDefinition() { |
| return MutableList.copyOf(stepsDefinition).asUnmodifiable(); |
| } |
| |
| transient Map<String,Pair<Integer,WorkflowStepDefinition>> stepsWithExplicitId; |
| transient List<WorkflowStepDefinition> stepsResolved; |
| @JsonIgnore |
| List<WorkflowStepDefinition> getStepsResolved() { |
| if (stepsResolved ==null) { |
| stepsResolved = MutableList.copyOf(WorkflowStepResolution.resolveSteps(getManagementContext(), WorkflowExecutionContext.this.stepsDefinition)); |
| } |
| return stepsResolved; |
| } |
| |
| @JsonIgnore |
| public Map<String, Pair<Integer,WorkflowStepDefinition>> getStepsWithExplicitIdById() { |
| if (stepsWithExplicitId ==null) stepsWithExplicitId = computeStepsWithExplicitIdById(getStepsResolved()); |
| return stepsWithExplicitId; |
| } |
| |
| |
| protected class Body implements Callable<Object> { |
| private WorkflowStepDefinition.ReplayContinuationInstructions continuationInstructions; |
| private Runnable intro = null; |
| private int stepsRun = 0; |
| |
| public Body() {} |
| |
| public Body(WorkflowStepDefinition.ReplayContinuationInstructions continuationInstructions) { |
| this.continuationInstructions = continuationInstructions; |
| } |
| |
| @Override |
| public String toString() { |
| return "WorkflowExecutionContext.Body["+workflowId+"; " + continuationInstructions + "]"; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| if (intro!=null) { |
| // intro needed to make dangling resumption wait |
| intro.run(); |
| } |
| return callWithLock(this::callSteps); |
| } |
| |
| protected Object callWithLock(Callable<Callable<Object>> handler) throws Exception { |
| AttributeSensor<String> lockSensor0 = null; |
| Entity lockEntity0 = null; |
| if (lock != null) { |
| String lockName = null; |
| if (lock instanceof String) lockName = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.WORKFLOW_INPUT, lock, TypeToken.of(String.class)); |
| else if (lock instanceof Map) { |
| lockName = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.WORKFLOW_INPUT, ((Map)lock).get("name"), TypeToken.of(String.class)); |
| |
| Object lockEntity00 = ((Map)lock).get("entity"); |
| if (lockEntity00!=null) { |
| lockEntity0 = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.WORKFLOW_INPUT, lockEntity00, TypeToken.of(Entity.class)); |
| log.debug(WorkflowExecutionContext.this + " using lock " + lockName + " on entity " + lockEntity0); |
| } |
| } |
| if (lockName==null) throw new IllegalArgumentException("Invalid lock object, should be a string or a map indicating a name and optional entity"); |
| lockSensor0 = Sensors.newStringSensor("lock-for-" + lockName); |
| if (lockEntity0==null) lockEntity0 = getEntity(); |
| } |
| AttributeSensor<String> lockSensor = lockSensor0; |
| Entity lockEntity = lockEntity0; |
| |
| AtomicBoolean mustClearLock = new AtomicBoolean(false); |
| if (lockSensor!=null) { |
| // acquire lock |
| // - step: set-sensor lock-for-count = ${workflow.id} |
| // require: |
| // any: |
| // - when: absent |
| // - equals: ${workflow.id} |
| // on-error: |
| // - retry backoff 50ms increasing 2x up to 5s |
| String wid = getWorkflowId(); |
| Duration delay = null; |
| String lastHolder = null; |
| while (true) { |
| AtomicReference<String> holder = new AtomicReference<>(); |
| lockEntity.sensors().modify(lockSensor, old -> { |
| if (old == null || old.equals(wid)) { |
| if (old != null) { |
| if (continuationInstructions != null) { |
| log.debug(WorkflowExecutionContext.this+" reasserting lock on " + lockSensor.getName() + " during replay"); |
| } else { |
| // i don't think this should be possible |
| log.warn("Entering block with lock on " + lockSensor.getName() + " when this workflow already holds the lock"); |
| } |
| } else { |
| log.debug(WorkflowExecutionContext.this+" acquired lock on " + lockSensor.getName()); |
| } |
| mustClearLock.set(true); |
| return Maybe.of(wid); |
| } |
| log.debug("Blocked by lock on " + lockSensor.getName() + ", currently held by " + old); |
| holder.set(old); |
| return Maybe.absent(); |
| }); |
| if (mustClearLock.get()) break; |
| // didn't get lock; probably do a retry |
| try { |
| if (delay==null || !Objects.equals(lastHolder, holder.get())) { |
| // reset initially, and if we observe the lock holder to change (highly competitive environment) |
| delay = Duration.millis(5); |
| } |
| Duration ddelay = delay; |
| Tasks.withBlockingDetails("Waiting for lock on " + lockSensor.getName() +" (held by workflow "+holder.get()+")", () -> { Time.sleep(ddelay); return null; }); |
| // increment delay for next time |
| delay = Duration.max(delay.multiply(1 + Math.random()), Duration.seconds(5)); |
| } catch (Exception e) { |
| throw Exceptions.propagate(e); |
| } |
| } |
| } |
| |
| Callable<Object> endHandler; |
| try { |
| endHandler = handler.call(); //super.doTaskBodyPossiblyWithLock(context, handler, isReplaying, continuationInstructions); |
| |
| } finally { |
| // clear if we set it, whether initially, on real replay, or on injected dangling failure |
| if (mustClearLock.get()) { |
| try { |
| DynamicTasks.waitForLast(); |
| } finally { |
| if (Entities.isUnmanagingOrNoLongerManaged(getEntity())) { |
| log.debug("Skipping clearance of lock on "+lockSensor.getName()+" in "+WorkflowExecutionContext.this+" because entity unmanaging here; expect auto-replay on resumption to pick up"); |
| } else { |
| Threads.runTemporarilyUninterrupted(() -> { |
| log.debug(WorkflowExecutionContext.this + " releasing lock on " + lockSensor.getName()); |
| ((EntityInternal.SensorSupportInternal) lockEntity.sensors()).remove(lockSensor); |
| }); |
| } |
| } |
| } |
| } |
| return endHandler.call(); |
| } |
| |
| boolean continueOnErrorHandledOrNextReplay; |
| AtomicReference<Boolean> timerCancelled; |
| |
| public Callable<Object> callSteps() throws Exception { |
| DynamicTasks.swallowChildrenFailures(); |
| |
| Task<?> timerTask = null; |
| timerCancelled = new AtomicReference<>(false); |
| try { |
| if (timeout != null) { |
| timerTask = initializeTimerFromWorkflowTimeout(timerTask); |
| } |
| |
| RecoveryAndReplay: do { |
| try { |
| boolean replaying = continuationInstructions!=null; |
| Integer replayFromStep = replaying ? continuationInstructions.stepToReplayFrom : null; |
| |
| if (!replaying) initializeWithoutContinuationInstructions(replayFromStep); |
| |
| continueOnErrorHandledOrNextReplay = false; |
| lastErrorHandlerOutput = null; |
| |
| WorkflowReplayUtils.updateOnWorkflowTaskStartupOrReplay(WorkflowExecutionContext.this, task, getStepsResolved(), !replaying, replayFromStep); |
| |
| // show task running |
| updateStatus(WorkflowStatus.RUNNING); |
| |
| if (replaying) initializeFromContinuationInstructions(replayFromStep); |
| |
| if (!Objects.equals(taskId, Tasks.current().getId())) |
| throw new IllegalStateException("Running workflow in unexpected task, " + taskId + " does not match " + task); |
| |
| int stepsConsidered = 0; |
| while (currentStepIndex >= 0 && currentStepIndex < getStepsResolved().size()) { |
| stepsConsidered++; |
| if (replaying && replayFromStep == null) { |
| // check step number and clear output before re-running |
| if (currentStepInstance==null || currentStepInstance.getStepIndex()!=currentStepIndex) { |
| throw new IllegalStateException("Running workflow at unexpected step, "+currentStepIndex+" v "+currentStepInstance); |
| } |
| currentStepInstance.output = null; |
| currentStepInstance.injectContext(WorkflowExecutionContext.this); |
| |
| log.debug("Replaying workflow '" + name + "', reusing instance " + currentStepInstance + " for step " + workflowStepReference(currentStepIndex) + ")"); |
| runCurrentStepInstanceApproved(getStepsResolved().get(currentStepIndex)); |
| |
| } else { |
| runCurrentStepIfPreconditions(); |
| } |
| |
| replaying = false; |
| |
| if (continuationInstructions!=null) { |
| continueOnErrorHandledOrNextReplay = true; |
| continue RecoveryAndReplay; |
| } |
| } |
| |
| log.debug("Completed workflow "+workflowId+" successfully; step count: "+stepsConsidered+" considered, "+stepsRun+" executed"); |
| |
| if (outputDefinition != null) { |
| output = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.STEP_FINISHING_POST_OUTPUT, outputDefinition, Object.class); |
| } else { |
| if (stepsRun>0) { |
| // (default is the output of the last step, if there is one, otherwise nothing) |
| // ((unlike steps, workflow output is not available as a default value, but previous step always is, so there is no need to do this |
| // before the above; slight chance if onError is triggered by a failure to resolve something in outputDefinition, but that can be ignored)) |
| output = getPreviousStepOutput(); |
| } |
| } |
| |
| updateOnSuccessfulCompletion(); |
| |
| } catch (Throwable e) { |
| try { |
| Pair<Throwable, WorkflowStatus> unhandledError = handleErrorAtWorkflow(e); |
| |
| if (unhandledError != null) { |
| return () -> endWithError(unhandledError.getLeft(), unhandledError.getRight()); |
| } |
| } catch (Throwable e2) { |
| // do not propagateIfFatal, we need to handle most throwables |
| log.debug("Uncaught error in workflow exception handler: "+ e2, e2); |
| return () -> endWithError(e2, WorkflowStatus.ERROR); |
| } |
| } |
| |
| } while (continueOnErrorHandledOrNextReplay); |
| |
| } finally { |
| if (timerTask != null && !timerTask.isDone() && !timerCancelled.get()) { |
| log.debug("Cancelling " + timerTask + " on completion of this task"); |
| timerTask.cancel(true); |
| } |
| } |
| |
| return this::endWithSuccess; |
| } |
| |
| private Pair<Throwable, WorkflowStatus> handleErrorAtWorkflow(Throwable e) { |
| boolean isTimeout = false; |
| |
| if (timerCancelled.get()) { |
| if (Exceptions.getCausalChain(e).stream().anyMatch(cause -> cause instanceof TimeoutException || cause instanceof InterruptedException || cause instanceof CancellationException || cause instanceof RuntimeInterruptedException)) { |
| // timed out, and a cause is related to cancellation |
| |
| // NOTE: this just gets used for logging, it is not ever stored by the calling task or returned to a user; |
| // when the task is cancelled, the only information available to callers is that it was cancelled |
| // (we could add indications of why but that is a lot of work, and not clear we need it) |
| TimeoutException timeoutException = new TimeoutException("Timeout after " + timeout + ": " + getName()); |
| timeoutException.initCause(e); |
| e = timeoutException; |
| isTimeout = true; |
| } |
| } |
| |
| WorkflowStatus provisionalStatus; |
| |
| if (Exceptions.isCausedByInterruptInAnyThread(e) || Exceptions.getFirstThrowableMatching(e, |
| // could do with a more precise check that workflow is cancelled or has timed out; exceptions could come from other unrelated causes (but unlikely) |
| t -> t instanceof CancellationException || t instanceof TimeoutException) != null) { |
| if (!Thread.currentThread().isInterrupted()) { |
| // might be a data model error |
| if (Exceptions.getFirstThrowableOfType(e, TemplateProcessor.TemplateModelDataUnavailableException.class) != null) { |
| provisionalStatus = WorkflowStatus.ERROR; |
| } else { |
| // cancelled or a subtask interrupted |
| provisionalStatus = WorkflowStatus.ERROR_CANCELLED; |
| } |
| } else { |
| provisionalStatus = WorkflowStatus.ERROR_CANCELLED; |
| } |
| if (provisionalStatus == WorkflowStatus.ERROR_CANCELLED) { |
| if (!getManagementContext().isRunning()) { |
| // if mgmt is shutting down we should record that. maybe enough if thread is interrupted we note that. |
| provisionalStatus = WorkflowStatus.ERROR_SHUTDOWN; |
| } else if (Entities.isUnmanagingOrNoLongerManaged(entity)) { |
| provisionalStatus = WorkflowStatus.ERROR_ENTITY_DESTROYED; |
| } |
| } |
| } else { |
| provisionalStatus = WorkflowStatus.ERROR; |
| } |
| |
| boolean errorHandled = false; |
| if (isTimeout) { |
| // don't run error handler |
| log.debug("Timeout in workflow '" + getName() + "' around step " + workflowStepReference(currentStepIndex) + ", throwing: " + Exceptions.collapseText(e)); |
| |
| } else if (Thread.currentThread().isInterrupted()) { |
| // don't run error handler |
| log.debug("Interrupt in workflow '" + getName() + "' around step " + workflowStepReference(currentStepIndex) + ", throwing: " + Exceptions.collapseText(e)); |
| |
| } else if (onError != null && (!(onError instanceof Collection) || !((Collection)onError).isEmpty())) { |
| try { |
| log.debug("Error in workflow '" + getName() + "' around step " + workflowStepReference(currentStepIndex) + ", running error handler"); |
| Task<WorkflowErrorHandling.WorkflowErrorHandlingResult> workflowErrorHandlerTask = WorkflowErrorHandling.createWorkflowErrorHandlerTask(WorkflowExecutionContext.this, task, e); |
| errorHandlerTaskId = workflowErrorHandlerTask.getId(); |
| WorkflowErrorHandling.WorkflowErrorHandlingResult result = DynamicTasks.queue(workflowErrorHandlerTask).getUnchecked(); |
| if (result != null) { |
| errorHandled = true; |
| |
| currentStepInstance.next = WorkflowReplayUtils.getNext(result.next, STEP_TARGET_NAME_FOR_END); |
| if (result.output != null) output = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.STEP_FINISHING_POST_OUTPUT, result.output, Object.class); |
| |
| moveToNextStep("Handled error in workflow around step " + workflowStepReference(currentStepIndex), result.next==null); |
| |
| if (continuationInstructions!=null || currentStepIndex < getStepsResolved().size()) { |
| continueOnErrorHandledOrNextReplay = true; |
| return null; |
| } |
| } // else errorHandled remains false and will fail below |
| |
| } catch (Exception e2) { |
| Throwable e0 = e; |
| if (Exceptions.getCausalChain(e2).stream().anyMatch(e3 -> e3==e0)) { |
| // wraps/rethrows original, don't need to log, but do return the new one |
| e = e2; |
| // } else if (Exceptions.isCausedByInterruptInAnyThread(e) && Exceptions.isCausedByInterruptInAnyThread(e2)) { |
| // // now handled above |
| // // if both are interrupted we can drop the trace, and return original; |
| // log.debug("Error where error handler was interrupted, after main thread was also interrupted: " + e2); |
| // log.trace("Full trace of original error was: " + e, e); |
| } else { |
| log.warn("Error in workflow '" + getName() + "' around step " + workflowStepReference(currentStepIndex) + " error handler for -- " + Exceptions.collapseText(e) + " -- threw another error (rethrowing): " + Exceptions.collapseText(e2)); |
| log.debug("Full trace of original error was: " + e, e); |
| e = e2; |
| } |
| } |
| |
| } else { |
| log.debug("Error in workflow '" + getName() + "' around step " + workflowStepReference(currentStepIndex) + ", no error handler so rethrowing: " + Exceptions.collapseText(e)); |
| } |
| |
| if (errorHandled) return null; |
| |
| if (replayAutomaticallyIfAppropriate(e)) return null; |
| |
| return Pair.of(e, provisionalStatus); |
| } |
| |
| private boolean replayAutomaticallyIfAppropriate(Throwable e) { |
| if (Boolean.TRUE.equals(replayableAutomatically) && Exceptions.getFirstThrowableOfType(e, DanglingWorkflowException.class)!=null) { |
| log.info("Automatic replay indicated for "+WorkflowExecutionContext.this+" when detected as dangling on server startup"); |
| |
| currentStepInstance.next = factory(true).makeInstructionsForReplayResuming("Replay resuming on dangling", false); |
| continueOnErrorHandledOrNextReplay = true; |
| return true; |
| } |
| |
| return false; |
| } |
| |
| private void updateOnSuccessfulCompletion() { |
| // finished -- checkpoint noting previous step and null for current because finished |
| updateStatus(WorkflowStatus.SUCCESS); |
| replayableLastStep = STEP_INDEX_FOR_END; |
| // record how it ended |
| oldStepInfo.compute(previousStepIndex == null ? STEP_INDEX_FOR_START : previousStepIndex, (index, old) -> { |
| if (old == null) old = new OldStepRecord(); |
| old.next = MutableSet.<Integer>of(STEP_INDEX_FOR_END).putAll(old.next); |
| old.nextTaskId = null; |
| return old; |
| }); |
| oldStepInfo.compute(STEP_INDEX_FOR_END, (index, old) -> { |
| if (old == null) old = new OldStepRecord(); |
| old.previous = MutableSet.<Integer>of(previousStepIndex == null ? STEP_INDEX_FOR_START : previousStepIndex).putAll(old.previous); |
| old.previousTaskId = previousStepTaskId; |
| return old; |
| }); |
| } |
| |
| private void resetWorkflowContextPreviousAndScratchVarsToStep(Integer step, boolean requireLastStep) { |
| if (step==null) { |
| // when resuming, keep them as they were set |
| return; |
| } |
| OldStepRecord last = oldStepInfo.get(step); |
| if (last != null) { |
| workflowScratchVariables = last.workflowScratch; |
| previousStepIndex = last.previous==null ? null : last.previous.stream().findFirst().orElse(null); |
| |
| } else { |
| if (requireLastStep) { |
| throw new IllegalStateException("Last step record required for step "+step+" to replay from there"); |
| } else { |
| // no such step; probably starting something which has never been run (not uncommon), or at a step which was never run (not sure if possible); |
| // in any case just keep the scratch vars as they were |
| } |
| } |
| // and ensure not null |
| if (workflowScratchVariables == null) workflowScratchVariables = MutableMap.of(); |
| } |
| |
| private void initializeFromContinuationInstructions(Integer replayFromStep) { |
| if (replayFromStep != null && replayFromStep == STEP_INDEX_FOR_START) { |
| log.debug("Replaying workflow '" + name + "', from start " + |
| "(was at " + (currentStepIndex == null ? "<UNSTARTED>" : workflowStepReference(currentStepIndex)) + ")"); |
| resetWorkflowContextPreviousAndScratchVarsToStep(replayFromStep, false); |
| currentStepIndex = 0; |
| |
| } else if (replayFromStep != null && replayFromStep == STEP_INDEX_FOR_END) { |
| log.debug("Replaying workflow '" + name + "', from end " + |
| "(was at " + (currentStepIndex == null ? "<UNSTARTED>" : workflowStepReference(currentStepIndex)) + ")"); |
| currentStepIndex = STEP_INDEX_FOR_END; |
| resetWorkflowContextPreviousAndScratchVarsToStep(replayFromStep, false); |
| currentStepInstance = null; |
| |
| } else { |
| // replaying workflow |
| log.debug("Replaying workflow '" + name + "', from step " + (replayFromStep == null ? "<CURRENT>" : workflowStepReference(replayFromStep)) + |
| " (was at " + (currentStepIndex == null ? "<UNSTARTED>" : workflowStepReference(currentStepIndex)) + ")"); |
| if (replayFromStep == null) { |
| // replayFromLast should correct the cases below |
| if (currentStepIndex == null) { |
| throw new IllegalStateException("Invalid instructions to continue from last bypassing convenience method, and there is no last"); |
| } else if (currentStepInstance == null || currentStepInstance.stepIndex != currentStepIndex) { |
| throw new IllegalStateException("Invalid instructions to continue from last step which is unknown, bypassing convenience method"); |
| } |
| } else { |
| currentStepIndex = replayFromStep; |
| } |
| // must reset, but okay if null, in that case we are continuing to a step which hasn't been initialized yet; use scratch vars |
| resetWorkflowContextPreviousAndScratchVarsToStep(currentStepIndex, false); |
| } |
| if (continuationInstructions.customWorkflowScratchVariables!=null) { |
| workflowScratchVariables.putAll(continuationInstructions.customWorkflowScratchVariables); |
| } |
| |
| } |
| |
| private void initializeWithoutContinuationInstructions(Integer replayFromStep) { |
| if (replayFromStep == null && currentStepIndex == null) { |
| currentStepIndex = 0; |
| log.debug("Starting workflow '" + name + "', moving to first step " + workflowStepReference(currentStepIndex)); |
| |
| } else if (replayFromStep==null && continueOnErrorHandledOrNextReplay) { |
| // workflow error handler indicated a next step to run |
| |
| } else { |
| // shouldn't come here |
| continueOnErrorHandledOrNextReplay = false; |
| throw new IllegalStateException("Should either be replaying or unstarted, but not invoked as replaying, and current=" + currentStepIndex + " replay=" + replayFromStep); |
| } |
| } |
| |
| private Task<?> initializeTimerFromWorkflowTimeout(Task<?> timerTask) { |
| Task<?> otherTask = Tasks.current(); |
| timerTask = Entities.submit(getEntity(), Tasks.builder().displayName("Timer for " + WorkflowExecutionContext.this.toString() + ":" + taskId) |
| .body(() -> { |
| try { |
| Time.sleep(timeout); |
| if (!otherTask.isDone()) { |
| timerCancelled.set(true); |
| log.debug("Cancelling " + otherTask + " after timeout of " + timeout); |
| otherTask.cancel(true); |
| } |
| } catch (Throwable e) { |
| if (Exceptions.isRootCauseIsInterruption(e)) { |
| // normal, just exit |
| } else { |
| throw Exceptions.propagate(e); |
| } |
| } |
| }).dynamic(false) |
| .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG) |
| .tag(BrooklynTaskTags.INESSENTIAL_TASK) |
| .build()); |
| return timerTask; |
| } |
| |
| private Object endWithSuccess() { |
| WorkflowReplayUtils.updateOnWorkflowSuccess(WorkflowExecutionContext.this, task, output); |
| persist(); |
| return output; |
| } |
| |
| private Object endWithError(Throwable e, WorkflowStatus provisionalStatus) { |
| updateStatus(provisionalStatus); |
| WorkflowReplayUtils.updateOnWorkflowError(WorkflowExecutionContext.this, task, e); |
| |
| try { |
| log.debug("Error running workflow " + this + "; will persist then rethrow: " + e); |
| log.trace("Error running workflow " + this + "; will persist then rethrow (details): " + e, e); |
| |
| } catch (Throwable e2) { |
| if (Entities.isUnmanagingOrNoLongerManaged(getEntity())) { |
| log.trace("Error persisting workflow (entity ending) " + this + " after error in workflow; persistence error (details): " + e2, e2); |
| } else { |
| log.error("Error persisting workflow " + this + " after error in workflow; persistence error: " + e2); |
| log.debug("Error persisting workflow " + this + " after error in workflow; persistence error (details): " + e2, e2); |
| log.warn("Error running workflow " + this + ", rethrowing without persisting because of persistence error (above): " + e); |
| } |
| log.trace("Error running workflow " + this + ", rethrowing without persisting because of persistence error (above): " + e, e); |
| } |
| |
| throw Exceptions.propagate(e); |
| } |
| |
| protected void runCurrentStepIfPreconditions() { |
| WorkflowStepDefinition step = getStepsResolved().get(currentStepIndex); |
| if (step!=null) { |
| currentStepInstance = new WorkflowStepInstanceExecutionContext(currentStepIndex, step, WorkflowExecutionContext.this); |
| DslPredicates.DslPredicate conditionResolved = step.getConditionResolved(currentStepInstance); |
| if (conditionResolved!=null) { |
| if (log.isTraceEnabled()) log.trace("Considering condition "+step.condition+" for "+ workflowStepReference(currentStepIndex)); |
| boolean conditionMet = DslPredicates.evaluateDslPredicateWithBrooklynObjectContext(conditionResolved, WorkflowExecutionContext.this, getEntityOrAdjunctWhereRunning()); |
| if (log.isTraceEnabled()) log.trace("Considered condition "+step.condition+" for "+ workflowStepReference(currentStepIndex)+": "+conditionMet); |
| if (!conditionMet) { |
| moveToNextStep("Skipping step "+ workflowStepReference(currentStepIndex), false); |
| return; |
| } |
| } |
| |
| // no condition or condition met -- record and run the step |
| |
| runCurrentStepInstanceApproved(step); |
| |
| } else { |
| // moving to floor/ceiling in treemap made sense when numero-ordered IDs are used, but not with list |
| throw new IllegalStateException("Cannot find step "+currentStepIndex); |
| } |
| } |
| |
| private void runCurrentStepInstanceApproved(WorkflowStepDefinition step) { |
| stepsRun++; |
| |
| Task<?> t; |
| if (continuationInstructions!=null) { |
| t = step.newTaskContinuing(currentStepInstance, continuationInstructions); |
| continuationInstructions = null; |
| } else { |
| t = step.newTask(currentStepInstance); |
| } |
| |
| // about to run -- checkpoint noting current and previous steps, and updating replayable from info |
| OldStepRecord currentStepRecord = oldStepInfo.compute(currentStepIndex, (index, old) -> { |
| if (old == null) old = new OldStepRecord(); |
| old.countStarted++; |
| if (!workflowScratchVariables.isEmpty()) |
| old.workflowScratch = MutableMap.copyOf(workflowScratchVariables); |
| else old.workflowScratch = null; |
| old.previous = MutableSet.<Integer>of(previousStepIndex == null ? STEP_INDEX_FOR_START : previousStepIndex).putAll(old.previous); |
| old.previousTaskId = previousStepTaskId; |
| old.nextTaskId = null; |
| return old; |
| }); |
| WorkflowReplayUtils.updateReplayableFromStep(WorkflowExecutionContext.this, step); |
| oldStepInfo.compute(previousStepIndex==null ? STEP_INDEX_FOR_START : previousStepIndex, (index, old) -> { |
| if (old==null) old = new OldStepRecord(); |
| old.next = MutableSet.<Integer>of(currentStepIndex).putAll(old.next); |
| old.nextTaskId = t.getId(); |
| return old; |
| }); |
| |
| errorHandlerContext = null; |
| errorHandlerTaskId = null; |
| currentStepInstance.next = null; // clear, eg if was set from a previous run; will be reset from step definition |
| // but don't clear output, in case a step is returning to itself and wants to reference previous_step.output |
| |
| persist(); |
| |
| BiConsumer<Object,Object> onFinish = (output,overrideNext) -> { |
| currentStepInstance.next = WorkflowReplayUtils.getNext(overrideNext, currentStepInstance, step); |
| if (output!=null) currentStepInstance.output = resolve(WorkflowExpressionResolution.WorkflowExpressionStage.STEP_FINISHING_POST_OUTPUT, output, Object.class); |
| }; |
| |
| // now run the step |
| try { |
| Duration duration = step.getTimeout(); |
| if (duration!=null) { |
| boolean isEnded = DynamicTasks.queue(t).blockUntilEnded(duration); |
| if (isEnded) { |
| currentStepInstance.output = t.getUnchecked(); |
| } else { |
| t.cancel(true); |
| throw new TimeoutException("Timeout after "+duration+": "+t.getDisplayName()); |
| } |
| } else { |
| currentStepInstance.output = DynamicTasks.queue(t).getUnchecked(); |
| } |
| |
| // allow output to be customized / overridden |
| onFinish.accept(step.output, null); |
| |
| } catch (Exception e) { |
| handleErrorAtStep(step, t, onFinish, e); |
| } |
| |
| oldStepInfo.compute(currentStepIndex, (index, old) -> { |
| if (old==null) { |
| log.warn("Lost old step info for "+this+", step "+index); |
| old = new OldStepRecord(); |
| } |
| old.countCompleted++; |
| // okay if this gets picked up by accident because we will check the stepIndex it records against the currentStepIndex, |
| // and ignore it if different |
| old.context = currentStepInstance; |
| return old; |
| }); |
| |
| previousStepTaskId = currentStepInstance.taskId; |
| previousStepIndex = currentStepIndex; |
| moveToNextStep("Completed step "+ workflowStepReference(currentStepIndex), false); |
| } |
| |
| private void handleErrorAtStep(WorkflowStepDefinition step, Task<?> stepTaskThrowingError, BiConsumer<Object, Object> onFinish, Exception error) { |
| WorkflowErrorHandling.handleErrorAtStep(getEntity(), step, currentStepInstance, stepTaskThrowingError, onFinish, error, null); |
| } |
| |
| private void logWarnOnExceptionOrDebugIfKnown(Exception e, String msg) { |
| WorkflowErrorHandling.logWarnOnExceptionOrDebugIfKnown(getEntity(), e, msg); |
| } |
| |
| private void moveToNextStep(String prefix, boolean inferredNext) { |
| prefix = prefix + "; "; |
| |
| Object specialNext = WorkflowReplayUtils.getNext(currentStepInstance); |
| |
| continuationInstructions = specialNext instanceof WorkflowStepDefinition.ReplayContinuationInstructions ? (WorkflowStepDefinition.ReplayContinuationInstructions) specialNext : null; |
| if (continuationInstructions!=null) { |
| log.debug(prefix + "proceeding to custom replay: "+continuationInstructions); |
| return; |
| } |
| |
| if (specialNext==null) { |
| currentStepIndex++; |
| if (currentStepIndex < getStepsResolved().size()) { |
| log.debug(prefix + "moving to sequential next step " + workflowStepReference(currentStepIndex)); |
| } else { |
| log.debug(prefix + "no further steps: Workflow completed"); |
| } |
| } else if (specialNext instanceof String) { |
| String explicitNext = (String)specialNext; |
| Maybe<Pair<Integer, Boolean>> nextResolved = getIndexOfStepId(explicitNext); |
| if (nextResolved.isAbsent()) { |
| log.warn(prefix + (inferredNext ? "inferred" : "explicit") + " next step '"+explicitNext+"' not found (failing)"); |
| // throw |
| nextResolved.get(); |
| } |
| if (nextResolved.get().getLeft()==null) { |
| throw new IllegalArgumentException("Next step '"+explicitNext+"' not supported here"); |
| } |
| |
| currentStepIndex = nextResolved.get().getLeft(); |
| if (nextResolved.get().getRight()) { |
| if (currentStepIndex < getStepsResolved().size()) { |
| log.debug(prefix + "moving to "+(inferredNext ? "inferred" : "explicit")+" next step " + workflowStepReference(currentStepIndex) + " for token '" + explicitNext + "'"); |
| } else { |
| log.debug(prefix + (inferredNext ? "inferred" : "explicit") + " next step '"+explicitNext+"': Workflow completed"); |
| } |
| } else { |
| log.debug(prefix + "moving to "+(inferredNext ? "inferred" : "explicit")+" next step " + workflowStepReference(currentStepIndex) + " for id '" + explicitNext + "'"); |
| } |
| } else { |
| throw new IllegalStateException("Illegal next definition: "+specialNext+" (type "+specialNext.getClass()+")"); |
| } |
| } |
| |
| String workflowStepReference(Integer index) { |
| if (index==null) return workflowId+"-<no-step>"; |
| |
| if (index>=getStepsResolved().size()) return getWorkflowStepReference(index, "<END>", false); |
| return getWorkflowStepReference(index, getStepsResolved().get(index)); |
| } |
| |
| public Body withIntro(Runnable intro) { |
| this.intro = intro; |
| return this; |
| } |
| } |
| |
| public Maybe<Pair<Integer,Boolean>> getIndexOfStepId(String next) { |
| if (next==null) return Maybe.absent("Null step ID supplied"); |
| Function<WorkflowExecutionContext, Integer> predefined = PREDEFINED_NEXT_TARGETS.get(next.toLowerCase()); |
| if (predefined!=null) return Maybe.of(Pair.of(predefined.apply(this), true)); |
| Pair<Integer, WorkflowStepDefinition> explicit = getStepsWithExplicitIdById().get(next); |
| if (explicit!=null) return Maybe.of(Pair.of(explicit.getLeft(), false)); |
| return Maybe.absent(new NoSuchElementException("Step with ID '"+next+"' not found")); |
| } |
| |
| String getWorkflowStepReference(int index, WorkflowStepDefinition step) { |
| return getWorkflowStepReference(index, step!=null ? step.id : null, false); |
| } |
| String getWorkflowStepReference(int index, String optionalStepId, boolean isError) { |
| // error handler step number not always available here, |
| // and risk of ID ambiguity with error-handler phrase; |
| // only for logging so this is okay; |
| // for canonical usage, prefer the Task-based method below |
| return workflowId+(index>=0 ? "-"+(index+1) : (index==STEP_INDEX_FOR_ERROR_HANDLER && isError) ? "" : "-"+indexCode(index)) |
| +(Strings.isNonBlank(optionalStepId) ? "-"+optionalStepId : "") |
| +(isError ? "-"+LABEL_FOR_ERROR_HANDLER : ""); |
| } |
| public String getWorkflowStepReference(Task<?> t) { |
| BrooklynTaskTags.WorkflowTaskTag wt = BrooklynTaskTags.getWorkflowTaskTag(t, false); |
| if (wt.getErrorHandlerIndex()!=null) { |
| // formula below not suitable for error tasks, but the name should be good |
| return t.getDisplayName(); |
| } |
| return wt.getWorkflowId() |
| +(wt.getStepIndex()!=null && wt.getStepIndex()>=0 && wt.getStepIndex()<stepsDefinition.size() ? "-"+(wt.getStepIndex()+1) : "") |
| +(wt.getErrorHandlerIndex()!=null ? "-error-handler-"+(wt.getErrorHandlerIndex()+1) : "") |
| ; |
| } |
| |
| private String indexCode(int index) { |
| // these numbers shouldn't be used for much, but they are used in a few places :( |
| if (index==STEP_INDEX_FOR_START) return STEP_TARGET_NAME_FOR_START; |
| if (index==STEP_INDEX_FOR_END) return STEP_TARGET_NAME_FOR_END; |
| if (index==STEP_INDEX_FOR_ERROR_HANDLER) return LABEL_FOR_ERROR_HANDLER; |
| return "neg-"+(index); // unknown |
| } |
| |
| } |
