| /* |
| * 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.google.common.base.Stopwatch; |
| import com.google.common.collect.Iterables; |
| import org.apache.brooklyn.api.entity.EntityLocal; |
| import org.apache.brooklyn.api.entity.EntitySpec; |
| import org.apache.brooklyn.api.mgmt.ManagementContext; |
| import org.apache.brooklyn.api.mgmt.Task; |
| import org.apache.brooklyn.api.sensor.AttributeSensor; |
| import org.apache.brooklyn.config.ConfigKey; |
| import org.apache.brooklyn.core.config.ConfigKeys; |
| import org.apache.brooklyn.core.entity.Entities; |
| import org.apache.brooklyn.core.entity.EntityAsserts; |
| import org.apache.brooklyn.core.entity.EntityInternal; |
| import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; |
| import org.apache.brooklyn.core.mgmt.internal.EntityManagementSupport; |
| import org.apache.brooklyn.core.mgmt.internal.LocalManagementContext; |
| import org.apache.brooklyn.core.mgmt.internal.ManagementContextInternal; |
| import org.apache.brooklyn.core.mgmt.rebind.RebindOptions; |
| import org.apache.brooklyn.core.mgmt.rebind.RebindTestFixture; |
| import org.apache.brooklyn.core.sensor.Sensors; |
| import org.apache.brooklyn.core.workflow.store.WorkflowStatePersistenceViaSensors; |
| import org.apache.brooklyn.entity.stock.BasicApplication; |
| import org.apache.brooklyn.test.Asserts; |
| import org.apache.brooklyn.test.ClassLogWatcher; |
| 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.task.DynamicTasks; |
| import org.apache.brooklyn.util.exceptions.Exceptions; |
| import org.apache.brooklyn.util.guava.Maybe; |
| import org.apache.brooklyn.util.text.Strings; |
| import org.apache.brooklyn.util.time.Duration; |
| import org.apache.brooklyn.util.time.Time; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.testng.annotations.Test; |
| |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.TimeoutException; |
| import java.util.function.Consumer; |
| import java.util.function.Supplier; |
| import java.util.stream.Collectors; |
| |
| public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicApplication> { |
| |
| private static final Logger log = LoggerFactory.getLogger(WorkflowPersistReplayErrorsTest.class); |
| |
| private BasicApplication app; |
| |
| @Override |
| protected LocalManagementContext decorateOrigOrNewManagementContext(LocalManagementContext mgmt) { |
| WorkflowBasicTest.addWorkflowStepTypes(mgmt); |
| return super.decorateOrigOrNewManagementContext(mgmt); |
| } |
| |
| @Override |
| protected BasicApplication createApp() { |
| return null; |
| } |
| |
| @Override protected BasicApplication rebind() throws Exception { |
| return rebind(RebindOptions.create().terminateOrigManagementContext(true)); |
| } |
| |
| Task<?> runStep(Object step, Consumer<BasicApplication> appFunction) { |
| return runSteps(MutableList.<Object>of(step), appFunction); |
| } |
| Task<?> runSteps(List<Object> steps, Consumer<BasicApplication> appFunction) { |
| return runSteps(steps, appFunction, null); |
| } |
| Task<?> runSteps(List<Object> steps, Consumer<BasicApplication> appFunction, ConfigBag initialEffectorConfig) { |
| BasicApplication app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| this.app = app; |
| return runStepsOnExistingApp("myWorkflow", steps, appFunction, initialEffectorConfig); |
| } |
| Task<?> runStepsOnExistingApp(String effectorName, List<Object> steps, Consumer<BasicApplication> appFunction, ConfigBag initialEffectorConfig) { |
| addEffector(effectorName, steps, appFunction, initialEffectorConfig); |
| |
| return app.invoke(app.getEntityType().getEffectorByName(effectorName).get(), null); |
| } |
| |
| private WorkflowEffector addEffector(String effectorName, List<Object> steps, Consumer<BasicApplication> appFunction, ConfigBag initialEffectorConfig) { |
| WorkflowEffector eff = new WorkflowEffector(ConfigBag.newInstance() |
| .configure(WorkflowEffector.EFFECTOR_NAME, effectorName) |
| .configure(WorkflowEffector.STEPS, steps) |
| .copy(initialEffectorConfig) |
| ); |
| if (appFunction !=null) appFunction.accept(app); |
| eff.apply((EntityLocal)app); |
| return eff; |
| } |
| |
| Task<?> lastInvocation; |
| WorkflowExecutionContext lastWorkflowContext; |
| |
| public final static List<Object> INCREMENTING_X_STEPS = MutableList.<Object>of( |
| "let integer x = ${entity.sensor.x} ?? 0", |
| "let x = ${x} + 1", |
| "set-sensor x = ${x}", |
| "wait ${entity.attributeWhenReady.gate}", |
| "let x = ${entity.sensor.x} + 10", |
| "set-sensor x = ${x}", |
| "return ${x}").asUnmodifiable(); |
| |
| private void runIncrementingX() { |
| lastInvocation = runSteps(INCREMENTING_X_STEPS, null); |
| |
| // ensure workflow sensor set immediately (this is done synchronously when effector invocation task is created, to ensure it can be resumed) |
| findSingleLastWorkflow(); |
| } |
| |
| private void runIncrementingXAwaitingGate() { |
| runIncrementingX(); |
| |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 1); |
| // refresh this for good measure (it won't have changed but feels like bad practice to rely on that) |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| } |
| |
| @Test |
| public void testWorkflowSensorValuesWhenPausedAndCanReplay() throws IOException { |
| runIncrementingXAwaitingGate(); |
| |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 1); |
| |
| Integer index = lastWorkflowContext.getCurrentStepIndex(); |
| Asserts.assertTrue(index >= 2 && index <= 3, "Index is "+index); |
| Asserts.assertEquals(lastWorkflowContext.status, WorkflowExecutionContext.WorkflowStatus.RUNNING); |
| Asserts.assertFalse(lastInvocation.isDone()); |
| |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| |
| lastInvocation.blockUntilEnded(Duration.seconds(2)); |
| Asserts.assertEquals(lastInvocation.getUnchecked(), 11); |
| Asserts.assertEquals(lastWorkflowContext.status, WorkflowExecutionContext.WorkflowStatus.SUCCESS); |
| |
| app.sensors().set(Sensors.newBooleanSensor("gate"), false); |
| Task<Object> invocation2 = DynamicTasks.submit(lastWorkflowContext.factory(false).createTaskReplaying(lastWorkflowContext.factory(false).makeInstructionsForReplayingFromStep(1, "Test", true)), app); |
| // sensor should go back to 1 because workflow vars are stored per-state |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 1); |
| Time.sleep(10); |
| Asserts.assertFalse(invocation2.isDone()); |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| invocation2.blockUntilEnded(Duration.seconds(2)); |
| Asserts.assertEquals(invocation2.getUnchecked(), 11); |
| } |
| |
| @Test |
| public void testWorkflowInterruptedAndCanReplay() throws IOException { |
| runIncrementingXAwaitingGate(); |
| |
| // cancel the workflow |
| lastWorkflowContext.getTask(true).get().cancel(true); |
| |
| // workflow should no longer proceed even if gate is set |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| |
| lastInvocation.blockUntilEnded(Duration.seconds(2)); |
| Asserts.assertTrue(lastInvocation.isError()); |
| |
| // sensor should not increment to 2 |
| // workflow should go to error (before invocation ended) |
| // TODO should show error when persisted |
| lastWorkflowContext.persist(); |
| Asserts.assertThat(lastWorkflowContext.status, status -> status.error); |
| // and sensor will not be set to 2 |
| EntityAsserts.assertAttributeEquals(app, Sensors.newSensor(Object.class, "x"), 1); |
| |
| Integer index = lastWorkflowContext.getCurrentStepIndex(); |
| Asserts.assertTrue(index >= 2 && index <= 3, "Index is "+index); |
| |
| Task<Object> invocation2 = DynamicTasks.submit(lastWorkflowContext.factory(false).createTaskReplaying(lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", true)), app); |
| // the gate is set so this will finish soon |
| Asserts.assertEquals(invocation2.getUnchecked(), 11); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newSensor(Object.class, "x"), 11); |
| } |
| |
| public static <T> T possiblyWithAutoFailAndReplay(boolean autoFailAndResume, Callable<T> callable) throws Exception { |
| Boolean old = null; |
| try { |
| if (!autoFailAndResume) { |
| old = EntityManagementSupport.AUTO_FAIL_AND_RESUME_WORKFLOWS; |
| EntityManagementSupport.AUTO_FAIL_AND_RESUME_WORKFLOWS = false; |
| } |
| return callable.call(); |
| } finally { |
| if (old!=null) { |
| EntityManagementSupport.AUTO_FAIL_AND_RESUME_WORKFLOWS = old; |
| } |
| } |
| } |
| |
| public static void possiblyWithAutoFailAndReplay(boolean autoFailAndResume, Runnable runnable) { |
| try { |
| possiblyWithAutoFailAndReplay(autoFailAndResume, () -> { |
| runnable.run(); |
| return null; |
| }); |
| } catch (Exception e) { |
| throw Exceptions.propagate(e); |
| } |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because slow, and tests variety of interruption points with randomised delay |
| public void testWorkflowShutdownAndCanReplay() throws Exception { |
| doTestWorkflowShutdownAndCanReplay(false); |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because slow, and tests variety of interruption points with randomised delay |
| public void testWorkflowShutdownAndAutomaticReplay() throws Exception { |
| doTestWorkflowShutdownAndCanReplay(true); |
| } |
| |
| public void doTestWorkflowShutdownAndCanReplay(boolean autoFailAndReplay) throws Exception { |
| runIncrementingX(); |
| |
| // variable delay, shouldn't matter when it is interrupted |
| Time.sleep((long) (Math.random()*Math.random()*200)); |
| |
| possiblyWithAutoFailAndReplay(autoFailAndReplay, () -> { |
| // do rebind |
| ManagementContext oldMgmt = mgmt(); |
| app = rebind(); |
| |
| // shutdown flags should get set on old object when destroyed |
| ((ManagementContextInternal) oldMgmt).terminate(); |
| WorkflowExecutionContext prevWorkflowContext = lastWorkflowContext; |
| Asserts.eventually(() -> prevWorkflowContext.status, status -> status.error); |
| Asserts.assertEquals(prevWorkflowContext.status, WorkflowExecutionContext.WorkflowStatus.ERROR_SHUTDOWN); |
| |
| // new object is present |
| lastWorkflowContext = findSingleLastWorkflow(); |
| |
| if (autoFailAndReplay) { |
| assertReplayedAfterRebindAndEventuallyThrowsDangling(lastWorkflowContext); |
| |
| } else { |
| // status should have been persisted as not ended |
| Asserts.assertEquals(lastWorkflowContext.status.ended, false); |
| } |
| |
| Asserts.assertThat(lastWorkflowContext.currentStepIndex, x -> x == null || x <= 3); |
| |
| // sensor might be one or might be null |
| Asserts.assertThat(app.sensors().get(Sensors.newSensor(Integer.class, "x")), x -> x == null || x == 1); |
| |
| // now we can tell it to resume from where it crashed |
| lastInvocation = Entities.submit(app, lastWorkflowContext.factory(false).createTaskReplaying(lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", true))); |
| |
| // will wait on gate, ie not finish |
| Time.sleep((long) (Math.random() * Math.random() * 200)); |
| Asserts.assertFalse(lastInvocation.isDone()); |
| |
| // workflow should now complete when gate is set |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| Asserts.assertEquals(lastInvocation.get(), 11); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newSensor(Object.class, "x"), 11); |
| return null; |
| }); |
| } |
| |
| private WorkflowExecutionContext findSingleLastWorkflow() { |
| Map<String, WorkflowExecutionContext> workflow = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app); |
| Asserts.assertSize(workflow, 1); |
| lastWorkflowContext = workflow.values().iterator().next(); |
| return lastWorkflowContext; |
| } |
| |
| private WorkflowExecutionContext findLastWorkflow(String workflowId) { |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).get(workflowId); |
| return Asserts.assertNotNull(lastWorkflowContext, "Cannot find workflow for "+workflowId); |
| } |
| |
| private void assertReplayedAfterRebindAndEventuallyThrowsDangling(WorkflowExecutionContext context) { |
| lastInvocation = mgmt().getExecutionManager().getTask(context.getTaskId()); |
| Asserts.assertNotNull(lastInvocation); |
| lastInvocation.blockUntilEnded(); |
| |
| // should get the dangling error |
| Asserts.assertEquals(context.status, WorkflowExecutionContext.WorkflowStatus.ERROR); |
| try { |
| lastInvocation.get(); |
| Asserts.shouldHaveFailedPreviously("Expected to throw "+DanglingWorkflowException.class); |
| } catch (Exception ex) { |
| // expected! |
| if (Exceptions.getFirstThrowableOfType(ex, DanglingWorkflowException.class)==null) { |
| throw new AssertionError("Wrong exception: "+ex, ex); |
| } |
| } |
| } |
| |
| @Test |
| public void testShutdownNotedIfManagementStopped() throws IOException { |
| runIncrementingXAwaitingGate(); |
| |
| Entities.destroyAll(mgmt()); |
| |
| lastInvocation.blockUntilEnded(Duration.seconds(5)); |
| Asserts.assertTrue(lastInvocation.isError()); |
| |
| // error is set, although it will not usually be persisted |
| Asserts.eventually(() -> lastWorkflowContext.status, status -> status.error); |
| if (lastWorkflowContext.status == WorkflowExecutionContext.WorkflowStatus.ERROR_SHUTDOWN || lastWorkflowContext.status == WorkflowExecutionContext.WorkflowStatus.ERROR_ENTITY_DESTROYED) { |
| // as expected |
| } else if (lastWorkflowContext.status == WorkflowExecutionContext.WorkflowStatus.ERROR) { |
| // sometimes happens; to be investigated |
| log.warn("Workflow ended with error, not error shutdown; value:\n"+lastInvocation.getStatusDetail(true)); |
| } else { |
| log.error("Workflow ended with wrong error status: "+lastWorkflowContext.status); |
| Asserts.fail("Workflow ended with wrong error status: "+lastWorkflowContext.status+ " / value:\n"+lastInvocation.getStatusDetail(true)); |
| } |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because a bit slow and non-deterministic |
| public void testNestedEffectorShutdownAndReplayedAutomatically() throws Exception { |
| doTestNestedWorkflowShutdownAndReplayed(true, "invoke-effector incrementXWithGate", app->{ |
| new WorkflowEffector(ConfigBag.newInstance() |
| .configure(WorkflowEffector.EFFECTOR_NAME, "incrementXWithGate") |
| .configure(WorkflowEffector.STEPS, INCREMENTING_X_STEPS) |
| ).apply((EntityLocal)app); |
| }); |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because a bit slow and non-deterministic |
| public void testNestedWorkflowShutdownAndReplayedAutomatically() throws Exception { |
| doTestNestedWorkflowShutdownAndReplayed(true, MutableMap.of("type", "workflow", "steps", INCREMENTING_X_STEPS), null); |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because a bit slow and non-deterministic |
| public void testNestedEffectorShutdownAndReplayedManually() throws Exception { |
| doTestNestedWorkflowShutdownAndReplayed(false, "invoke-effector incrementXWithGate", app->{ |
| new WorkflowEffector(ConfigBag.newInstance() |
| .configure(WorkflowEffector.EFFECTOR_NAME, "incrementXWithGate") |
| .configure(WorkflowEffector.STEPS, MutableList.<Object>of("log running nested incrementX in ${workflow.name}").appendAll(INCREMENTING_X_STEPS)) |
| ).apply((EntityLocal)app); |
| }); |
| } |
| |
| @Test(groups="Integration", invocationCount = 10) // because a bit slow and non-deterministic |
| public void testNestedWorkflowShutdownAndReplayedManually() throws Exception { |
| doTestNestedWorkflowShutdownAndReplayed(false, MutableMap.of("type", "workflow", "steps", INCREMENTING_X_STEPS), null); |
| } |
| |
| void doTestNestedWorkflowShutdownAndReplayed(boolean autoFailAndReplay, Object call, Consumer<BasicApplication> initializer) throws Exception { |
| lastInvocation = runSteps(MutableList.<Object>of( |
| "let y = ${entity.sensor.y} ?? 0", |
| "let y = ${y} + 1", |
| "set-sensor y = ${y}", |
| call, |
| "let x = ${workflow.previous_step.output}", |
| "let y = ${y} + 10", |
| "set-sensor y = ${y}", |
| "let z = ${y} * 100 + ${x}", |
| "return ${z}" |
| ), initializer); |
| |
| // run once with no interruption, make sure fine |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| Asserts.assertEquals(lastInvocation.get(), 1111); |
| app.sensors().set(Sensors.newBooleanSensor("gate"), false); |
| |
| // now invoke again |
| lastInvocation = app.invoke(app.getEntityType().getEffectorByName("myWorkflow").get(), null); |
| String workflowId = BrooklynTaskTags.getWorkflowTaskTag(lastInvocation, false).getWorkflowId(); |
| |
| // interrupt at any point, probably when gated |
| Time.sleep((long) (Math.random()*Math.random()*200)); |
| ManagementContext oldMgmt = mgmt(); |
| |
| possiblyWithAutoFailAndReplay(autoFailAndReplay, () -> { |
| app = rebind(); |
| ((ManagementContextInternal) oldMgmt).terminate(); |
| |
| lastWorkflowContext = findLastWorkflow(lastInvocation.getId()); |
| |
| if (autoFailAndReplay) { |
| assertReplayedAfterRebindAndEventuallyThrowsDangling(lastWorkflowContext); |
| |
| } else { |
| // status should have been persisted as not ended |
| Asserts.assertEquals(lastWorkflowContext.status.ended, false); |
| } |
| |
| lastInvocation = Entities.submit(app, lastWorkflowContext.factory(false).createTaskReplaying(lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", true))); |
| if (lastInvocation.blockUntilEnded(Duration.millis(20))) { |
| Asserts.fail("Invocation ended when it shouldn't have, with "+lastInvocation.get()); |
| } |
| |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| Asserts.assertEquals(lastInvocation.get(), 2222); |
| |
| // wait for effector to be invoked |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 22); |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "y"), 22); |
| return null; |
| }); |
| } |
| |
| public static List<Object> INCREMENTING_X_STEPS_SETTING_REPLAYABLE(boolean replayableOnBuggyStep1, boolean replayableOnGoodStep2) { |
| return MutableList.<Object>of( |
| MutableMap.of("s", "let integer x = ${entity.sensor.x} ?? 0", "replayable", replayableOnBuggyStep1 ? "from here" : null), |
| MutableMap.of("s", "let x = ${x} + 1", "replayable", replayableOnGoodStep2 ? "from here" : null), |
| "set-sensor x = ${x}", |
| "wait ${entity.attributeWhenReady.gate}", |
| "return ${x}").asUnmodifiable(); |
| } |
| |
| private void runIncrementingXSettingReplayable(boolean replayableOnBuggyStep1, boolean replayableOnGoodStep2) { |
| lastInvocation = runSteps(INCREMENTING_X_STEPS_SETTING_REPLAYABLE(replayableOnBuggyStep1, replayableOnGoodStep2), null); |
| |
| // ensure workflow sensor set immediately (this is done synchronously when effector invocation task is created, to ensure it can be resumed) |
| findSingleLastWorkflow(); |
| } |
| |
| private void runIncrementingXAwaitingGateSettingReplayable(boolean replayableOnBuggyStep1, boolean replayableOnGoodStep2) { |
| runIncrementingXSettingReplayable(replayableOnBuggyStep1, replayableOnGoodStep2); |
| |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 1); |
| // refresh this for good measure (it won't have changed but feels like bad practice to rely on that) |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testWorkflowInterruptedAndCanReplaySettingReplayableBuggy() throws IOException { |
| doTestWorkflowInterruptedAndCanReplaySettingReplayable(true, false, false, true); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testWorkflowInterruptedAndCanReplaySettingReplayableSensibleReplayPoint() throws IOException { |
| doTestWorkflowInterruptedAndCanReplaySettingReplayable(false, true, false, false); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testWorkflowInterruptedAndCanReplaySettingReplayableSensibleResuming() throws IOException { |
| doTestWorkflowInterruptedAndCanReplaySettingReplayable(false, false, true, false); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testWorkflowInterruptedAndCanReplaySettingReplayableSensibleResumingEvenIfBuggyReplayPoint() throws IOException { |
| doTestWorkflowInterruptedAndCanReplaySettingReplayable(true, false, true, false); |
| } |
| |
| void doTestWorkflowInterruptedAndCanReplaySettingReplayable(boolean replayableOnBuggyStep1, boolean replayableOnGoodStep2, boolean isResuming, boolean expectBuggy) throws IOException { |
| runIncrementingXAwaitingGateSettingReplayable(replayableOnBuggyStep1, replayableOnGoodStep2); |
| |
| // cancel the workflow |
| lastWorkflowContext.getTask(true).get().cancel(true); |
| lastInvocation.blockUntilEnded(Duration.seconds(2)); |
| |
| // in buggy mode, should want to continue from 0 |
| int expected = -1; |
| if (expectBuggy) { |
| Asserts.assertThat(lastWorkflowContext.replayableLastStep, step -> step == 0); |
| expected = 2; |
| } else { |
| if (!isResuming) { |
| Asserts.assertThat(lastWorkflowContext.replayableLastStep, step -> step == 1); |
| } |
| expected = 1; |
| } |
| |
| // replay |
| Task<Object> invocation2 = DynamicTasks.submit(lastWorkflowContext.factory(false).createTaskReplaying( |
| isResuming |
| ? lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", false) |
| : lastWorkflowContext.factory(false).makeInstructionsForReplayingFromLastReplayable("test", false) |
| ), app); |
| |
| // should get 2 because it replays from 0 |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| Asserts.assertEquals(invocation2.getUnchecked(), expected); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newSensor(Object.class, "x"), expected); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testNestedWorkflowInterruptedAndCanReplaySettingReplayable_BuggyNestedReplay() throws IOException { |
| doTestNestedEffectorWorkflowInterruptedAndCanReplaySettingReplayable(true, true, true, false, false, |
| 1, 0, true); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testNestedWorkflowInterruptedAndCanReplaySettingReplayable_BuggyTopLevelReplay() throws IOException { |
| doTestNestedEffectorWorkflowInterruptedAndCanReplaySettingReplayable(true, false,false, false, false, |
| -1, null, true); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testNestedWorkflowInterruptedAndCanReplaySettingReplayable_AllPointsAreBuggy() throws IOException { |
| doTestNestedEffectorWorkflowInterruptedAndCanReplaySettingReplayable(true, true, true, true, false, |
| 1, 1, true); |
| } |
| |
| @Test(groups="Integration") //because slow |
| public void testNestedWorkflowInterruptedAndCanReplaySettingReplayable_SensibleResuming() throws IOException { |
| doTestNestedEffectorWorkflowInterruptedAndCanReplaySettingReplayable(true, true, true, false, true, |
| 1, 0, false); |
| } |
| |
| void doTestNestedEffectorWorkflowInterruptedAndCanReplaySettingReplayable(boolean replayableFromStartOuter, boolean replayableOnEffector, |
| boolean replayableOnBuggyStep1, boolean replayableOnGoodStep2, boolean isResuming, |
| Integer expectedOuterStep, Integer expectedNestedStep, boolean expectBuggy) throws IOException { |
| // replay is only "good" if we set a replay point _after_ the sensor was read; |
| // otherwise it increments x twice |
| lastInvocation = runSteps( |
| MutableList.of( |
| "sleep 20ms", |
| MutableMap.of("s", "invoke-effector nestedWorkflow", "replayable", replayableOnEffector ? "from here" : null)), |
| app -> { |
| List<Object> steps = INCREMENTING_X_STEPS_SETTING_REPLAYABLE(replayableOnBuggyStep1, replayableOnGoodStep2); |
| |
| WorkflowEffector eff = new WorkflowEffector(ConfigBag.newInstance() |
| .configure(WorkflowEffector.EFFECTOR_NAME, "nestedWorkflow") |
| .configure(WorkflowEffector.STEPS, steps) |
| ); |
| eff.apply((EntityLocal)app); |
| }, |
| ConfigBag.newInstance().configure(WorkflowEffector.REPLAYABLE, replayableFromStartOuter ? "from start" : null)); |
| |
| // ensure workflow sensor set immediately (this is done synchronously when effector invocation task is created, to ensure it can be resumed) |
| findSingleLastWorkflow(); |
| |
| EntityAsserts.assertAttributeEqualsEventually(app, Sensors.newSensor(Object.class, "x"), 1); |
| Time.sleep(Duration.millis(50)); // give it 50ms to make sure it gets to the waiting step |
| // refresh this for good measure (it won't have changed but feels like bad practice to rely on that) |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| |
| // cancel the workflow |
| lastWorkflowContext.getTask(true).get().cancel(true); |
| lastInvocation.blockUntilEnded(Duration.seconds(2)); |
| |
| // in buggy mode, should want to continue from 0 |
| int expected = -1; |
| Asserts.assertThat(lastWorkflowContext.replayableLastStep, step -> Objects.equals(step, expectedOuterStep)); |
| |
| Map<String, WorkflowExecutionContext> workflows = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app); |
| Asserts.assertSize(workflows, 2); |
| WorkflowExecutionContext nestedWorkflow = workflows.values().stream().filter(w -> w.getParentTag() != null).findAny().get(); |
| |
| Asserts.assertThat(nestedWorkflow.replayableLastStep, step -> Objects.equals(step, expectedNestedStep)); |
| |
| if (expectBuggy) { |
| expected = 2; |
| } else { |
| expected = 1; |
| } |
| |
| // replay |
| Task<Object> invocation2 = DynamicTasks.submit(lastWorkflowContext.factory(false).createTaskReplaying( |
| isResuming |
| ? lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", false) |
| : lastWorkflowContext.factory(false).makeInstructionsForReplayingFromLastReplayable("test", false) |
| ), app); |
| |
| // should get 2 because it replays from 0 |
| app.sensors().set(Sensors.newBooleanSensor("gate"), true); |
| Asserts.assertEquals(invocation2.getUnchecked(), expected); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newSensor(Object.class, "x"), expected); |
| } |
| |
| @Test |
| public void testSimpleErrorHandlerOnStep() throws IOException { |
| try (ClassLogWatcher logWatcher = new ClassLogWatcher(getClass().getPackage().getName())) { |
| lastInvocation = runSteps(MutableList.of( |
| MutableMap.of("s", "invoke-effector does-not-exist", |
| "output", "should have failed", |
| "on-error", MutableList.of( |
| MutableMap.of("type", "no-op", |
| "output", "error-handler worked!")))), |
| null); |
| Asserts.assertEquals(lastInvocation.getUnchecked(), "error-handler worked!"); |
| |
| List<String> msgs = logWatcher.getMessages().stream().filter(x -> !x.startsWith("Blocked by lock")).collect(Collectors.toList()); |
| // can have "Blocked by lock on lock-for-incrementor, currently held by JPuhvC9I" from a previous invocation? |
| |
| log.info("Error handler output:\n"+msgs.stream().collect(Collectors.joining("\n"))); |
| Asserts.assertEntriesSatisfy(msgs, MutableList.of( |
| m -> m.matches("Starting workflow 'myWorkflow .workflow effector.', moving to first step .*-1"), |
| m -> m.matches("Starting step .*-1 in task .*"), |
| m -> m.matches("Encountered error in step .*-1 '1 - invoke-effector does-not-exist' .handler present.: No effector matching 'does-not-exist'"), |
| m -> m.matches("Starting .*-1-error-handler with 1 step in task .*"), |
| m -> m.matches("Starting .*-1-error-handler-1 in task .*"), |
| m -> m.matches("Completed handler .*-1-error-handler; no next step indicated so proceeding to default next step"), |
| m -> m.matches("Completed step .*-1; no further steps: Workflow completed"), |
| m -> m.matches("Completed workflow .* successfully; step count: 1 considered, 1 executed") )); |
| |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| Asserts.assertEquals(lastWorkflowContext.currentStepIndex, (Integer) 1); |
| } |
| } |
| |
| @Test |
| public void testSimpleErrorHandlerOnWorkflow() throws IOException { |
| try (ClassLogWatcher logWatcher = new ClassLogWatcher(getClass().getPackage().getName())) { |
| lastInvocation = runSteps(MutableList.of( |
| MutableMap.of("s", "invoke-effector does-not-exist", |
| "output", "should have failed"), |
| "log should not run"), |
| null, |
| ConfigBag.newInstance().configure( |
| WorkflowEffector.ON_ERROR, MutableList.of( |
| MutableMap.of("type", "no-op", |
| "output", "error-handler worked!")))); |
| Asserts.assertEquals(lastInvocation.getUnchecked(), "error-handler worked!"); |
| |
| List<String> msgs = logWatcher.getMessages().stream().filter(x -> !x.startsWith("Blocked by lock")).collect(Collectors.toList()); |
| // can have "Blocked by lock on lock-for-incrementor, currently held by JPuhvC9I" from a previous invocation? |
| |
| log.info("Error handler output:\n"+msgs.stream().collect(Collectors.joining("\n"))); |
| Asserts.assertEntriesSatisfy(msgs, MutableList.of( |
| m -> m.matches("Starting workflow 'myWorkflow .workflow effector.', moving to first step .*-1"), |
| m -> m.matches("Starting step .*-1 in task .*"), |
| m -> m.matches("myWorkflow .*: Error in step '1 - invoke-effector does-not-exist'; rethrowing: No effector matching 'does-not-exist'"), |
| m -> m.matches("Error in workflow 'myWorkflow .workflow effector.' around step .*-1, running error handler"), |
| m -> m.matches("Encountered error in workflow .*/.* 'myWorkflow' .handler present.: No effector matching 'does-not-exist'"), |
| m -> m.matches("Starting .*-error-handler with 1 step in task .*"), |
| m -> m.matches("Starting .*-error-handler-1 in task .*"), |
| m -> m.matches("Completed handler .*-error-handler; no next step indicated so proceeding to default next step"), |
| m -> m.matches("Handled error in workflow around step .*-1; inferred next step 'end': Workflow completed"))); |
| } |
| } |
| |
| @Test |
| public void testSimpleErrorHandlerOnWorkflowFailing() throws IOException { |
| lastInvocation = runSteps(MutableList.of("invoke-effector does-not-exist"), |
| null, |
| ConfigBag.newInstance().configure( |
| WorkflowEffector.ON_ERROR, MutableList.of("set-sensor had_error = yes", "fail rethrow message rethrown")) ); |
| Asserts.assertFailsWith(() -> lastInvocation.getUnchecked(), e -> { |
| Asserts.expectedFailureContains(e, "rethrown"); |
| Asserts.assertThat(Exceptions.getCausalChain(e), ee -> ee.stream().filter(e2 -> e2.toString().contains("does-not-exist")).findAny().isPresent()); |
| return true; |
| }); |
| Asserts.assertEquals(app.sensors().get(Sensors.newSensor(Object.class, "had_error")), "yes"); |
| findSingleLastWorkflow(); |
| Asserts.assertEquals(lastWorkflowContext.status, WorkflowExecutionContext.WorkflowStatus.ERROR); |
| } |
| |
| @Test |
| public void testErrorHandlerListWithGotoExit() throws IOException { |
| try (ClassLogWatcher logWatcher = new ClassLogWatcher(getClass().getPackage().getName())) { |
| lastInvocation = runSteps(MutableList.of( |
| MutableMap.of("s", "invoke-effector does-not-exist", |
| "output", "NOT returned", |
| "on-error", MutableList.of( |
| "log Error handler 1-1", |
| MutableMap.of("step", "log Error handler 1-2", |
| "output", "from 1-2"), |
| MutableMap.of("step", "log NOT shown because of condition", |
| "condition", MutableMap.of("target", "${output}", "equals", "NOT matched")), |
| MutableMap.of("step", "log Error handler 1-4 has output ${output}", |
| "condition", MutableMap.of("target", "${output}", "glob", "from *")), |
| MutableMap.of("step", "log Step created-but-not-logged because of bad variable ${not_available}", |
| "on-error", MutableList.of( |
| MutableMap.of("step", "log Error handler 1-5-1", "output", "from 1-5-1"), |
| "goto exit", |
| "log NOT shown after inner exit") |
| ), |
| "log Error handler 1-6", |
| "goto exit", |
| "log NOT shown because of earlier exit") |
| ), |
| "log Step 2 has output ${output}" |
| ), |
| null); |
| Asserts.assertEquals(lastInvocation.getUnchecked(), "from 1-5-1"); |
| |
| List<String> msgs = logWatcher.getMessages(); |
| log.info("Error handler output:\n"+msgs.stream().collect(Collectors.joining("\n"))); |
| Asserts.assertEquals(msgs.stream().filter(s -> s.contains("NOT")).findAny().orElse(null), null); |
| Asserts.assertEquals(msgs.stream().filter(s -> s.contains("created-but-not-logged") && !s.contains("Creating handler")).findAny().orElse(null), null); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("1-1")).findAny().orElse(null)); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("1-2")).findAny().orElse(null)); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("1-4")).findAny().orElse(null)); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("1-5-1")).findAny().orElse(null)); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("1-6")).findAny().orElse(null)); |
| Asserts.assertNotNull(msgs.stream().filter(s -> s.contains("Step 2 has output from 1-5-1")).findAny().orElse(null)); |
| } |
| } |
| |
| @Test |
| public void testErrorHandlerRethrows() throws IOException { |
| lastInvocation = runSteps(MutableList.of( |
| MutableMap.of("step", "fail message expected exception", |
| "output", "should have failed", |
| "on-error", MutableList.of( |
| MutableMap.of("step", "return not applicable", |
| "condition", "not matched")))), |
| null); |
| Asserts.assertFailsWith(() -> Asserts.fail("Did not fail, returned: "+lastInvocation.getUnchecked()), |
| e -> Asserts.expectedFailureContainsIgnoreCase(e, "expected exception")); |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| Asserts.assertEquals(lastWorkflowContext.currentStepIndex, (Integer) 0); |
| } |
| |
| @Test |
| public void testMultilineErrorRegex() throws IOException { |
| lastInvocation = runSteps(MutableList.of( |
| MutableMap.of("step", "log ${var_does_not_exist}", |
| "output", "should have failed", |
| "on-error", MutableList.of( |
| MutableMap.of("step", "return error handled", |
| "condition", MutableMap.of("regex", ".*InvalidReference.*var_does_not_exist.*"))))), |
| null); |
| Asserts.assertEquals(lastInvocation.getUnchecked(), "error handled"); |
| } |
| |
| @Test |
| public void testTimeoutOnStep() throws Exception { |
| doTestTimeout(false, true); |
| } |
| |
| @Test |
| public void testTimeoutOnWorkflow() throws Exception { |
| doTestTimeout(false, false); |
| } |
| |
| @Test |
| public void testTimeoutNotExceededOnStep() throws Exception { |
| doTestTimeout(true, true); |
| } |
| |
| @Test |
| public void testTimeoutNotExceededOnWorkflow() throws Exception { |
| doTestTimeout(true, false); |
| } |
| |
| public void doTestTimeout(boolean finishesBeforeTimeout, boolean onStep) throws Exception { |
| Stopwatch sw = Stopwatch.createStarted(); |
| Duration sleepTime, timeout; |
| ConfigBag effConfig = ConfigBag.newInstance(); |
| if (finishesBeforeTimeout) { |
| sleepTime = Duration.seconds(1); |
| timeout = Duration.seconds(10); |
| } else { |
| sleepTime = Duration.seconds(10); |
| timeout = Duration.seconds(1); |
| } |
| |
| Map<String,Object> step = MutableMap.of("s", "sleep "+sleepTime); |
| if (!onStep) effConfig.configure(WorkflowEffector.TIMEOUT, timeout); |
| else step.put("timeout", ""+timeout); |
| |
| try { |
| lastInvocation = runSteps(MutableList.of(step), null, effConfig); |
| Object result = lastInvocation.getUnchecked(); |
| |
| if (finishesBeforeTimeout) { |
| // expected |
| } else { |
| throw Asserts.fail("Should have timed out, instead gave: " + result); |
| } |
| |
| } catch (Exception e) { |
| if (!finishesBeforeTimeout && Exceptions.getFirstThrowableOfType(e, onStep ? (Class)TimeoutException.class : CancellationException.class)!=null) { |
| // expected; for step, the workflow should always capture it as a timeout exception; |
| // for workflow-level timeouts, per comments in WorkflowExecutionContext where it throws TimeoutException, |
| // we cannot easily prevent a CancellationException from being reported |
| } else { |
| throw Asserts.fail(e); |
| } |
| } |
| |
| if (finishesBeforeTimeout) { |
| if (Duration.of(sw).isShorterThan(sleepTime)) |
| Asserts.fail("Finished too quick: " + Duration.of(sw)); |
| if (Duration.of(sw).isLongerThan(timeout)) Asserts.fail("Took too long: " + Duration.of(sw)); |
| } else { |
| if (Duration.of(sw).isShorterThan(timeout)) Asserts.fail("Finished too quick: " + Duration.of(sw)); |
| if (Duration.of(sw).isLongerThan(sleepTime)) Asserts.fail("Took too long: " + Duration.of(sw)); |
| } |
| |
| //app.getManagementContext().getExecutionManager().getTasksWithAllTags( |
| Asserts.eventually(() -> ((EntityInternal)app).getExecutionContext().getTasks().stream().filter(t -> !t.isDone()).collect(Collectors.toList()), |
| unfinishedTasks -> { |
| System.out.println("TASKS: "+unfinishedTasks); |
| System.out.println(lastInvocation); |
| return unfinishedTasks.isEmpty(); |
| }, Duration.FIVE_SECONDS |
| ); |
| } |
| |
| @Test |
| public void testFailAndErrorHandlerAsListOrMapOrString() { |
| MutableList<MutableMap<String, Serializable>> errorHandler = MutableList.of(MutableMap.of("step", "return Yay WTF", |
| "condition", MutableMap.of("error-cause", MutableMap.of("regex", ".*Fail.*wtf.*")))); |
| doTestFail(errorHandler); |
| doTestFail(errorHandler.get(0)); |
| doTestFail(errorHandler.get(0).get("step")); |
| } |
| |
| void doTestFail(Object errorHandler) { |
| Task<?> out = runSteps( |
| MutableList.of("fail message wtf"), |
| null, |
| ConfigBag.newInstance().configure(WorkflowCommonConfig.ON_ERROR, errorHandler) |
| ); |
| Asserts.assertEquals(out.getUnchecked(), "Yay WTF"); |
| } |
| |
| @Test |
| public void testReplayableDisabled() { |
| Task<?> out = runSteps( |
| MutableList.of("workflow replayable from here", "let x = ${entity.sensor.count} + 1 ?? 1", "set-sensor count = ${x}", "fail message wtf"), |
| null, |
| ConfigBag.newInstance().configure(WorkflowCommonConfig.REPLAYABLE, "disabled")); |
| out.blockUntilEnded(); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newIntegerSensor("count"), 1); |
| |
| lastWorkflowContext = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).values().iterator().next(); |
| |
| WorkflowExecutionContext.Factory wff = lastWorkflowContext.factory(true); |
| DynamicTasks.submit(wff.createTaskReplaying(wff.makeInstructionsForReplayingFromLastReplayable("testing", false)), app).blockUntilEnded(); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newIntegerSensor("count"), 2); |
| |
| WorkflowExecutionContext.Factory wfd = lastWorkflowContext.factory(false); |
| Asserts.assertFailsWith(() -> wfd.createTaskReplaying(wfd.makeInstructionsForReplayingFromLastReplayable("testing", false)), |
| e -> { |
| Asserts.expectedFailureContainsIgnoreCase(e, "disabled"); |
| return true; |
| }); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newIntegerSensor("count"), 2); |
| |
| DynamicTasks.submit(wfd.createTaskReplaying(wfd.makeInstructionsForReplayingFromLastReplayable("testing", true)), app).blockUntilEnded(); |
| EntityAsserts.assertAttributeEquals(app, Sensors.newIntegerSensor("count"), 3); |
| } |
| |
| @Test |
| public void testRetentionQuickly() throws Exception { |
| app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| |
| // always use the latest mgmt context! |
| Supplier<WorkflowStatePersistenceViaSensors> wp = () -> new WorkflowStatePersistenceViaSensors(mgmt()); |
| BrooklynTaskTags.WorkflowTaskTag w1, w2, w3, w4; |
| |
| w1 = doTestRetentionDisabled("context", "min(1,2) hash my-fixed-hash", false, false, false); |
| Asserts.assertEquals(lastWorkflowContext.getRetentionSettings().expiryResolved, "min(1,2)"); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId())); |
| |
| w1 = doTestRetentionDisabled(2, "hash my-fixed-hash min(1,context)", false, false, false); |
| Asserts.assertEquals(lastWorkflowContext.getRetentionSettings().expiryResolved, "min(1,2)"); |
| |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId())); |
| |
| // invoking our test gives a new workflow hash because the effector name is different |
| w2 = doTestRetentionDisabled(2, "1", false, false, false); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId())); |
| |
| // reinvoking that effector still gives 2 |
| Task<?> t = app.invoke(app.getEntityType().getEffectorByName("myWorkflow" + effNameCount).get(), null); |
| t.blockUntilEnded(); |
| w2 = BrooklynTaskTags.getWorkflowTaskTag(t, false); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId())); |
| |
| // hash accepts variables |
| app.config().set(ConfigKeys.newStringConfigKey("hash"), "my-fixed-hash"); |
| |
| // this hash replaces old w1 |
| w1 = doTestRetentionDisabled("context", "min(1,2) hash ${entity.config.hash}", false, false, false); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId())); // should replace the one above |
| |
| // workflow.id as the hash variable means each invocations has its own retention |
| w3 = doTestRetentionDisabled("context", "1 hash ${workflow.id}", false, false, false); |
| |
| t = app.invoke(app.getEntityType().getEffectorByName("myWorkflow" + effNameCount).get(), null); |
| t.blockUntilEnded(); |
| w4 = BrooklynTaskTags.getWorkflowTaskTag(t, false); |
| |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId(), w3.getWorkflowId(), w4.getWorkflowId())); // should replace the one above |
| } |
| |
| @Test(groups="Integration") // very slow |
| public void testRetentionManyWaysIncludingDisabled() throws Exception { |
| app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| |
| // always use the latest mgmt context! |
| Supplier<WorkflowStatePersistenceViaSensors> wp = () -> new WorkflowStatePersistenceViaSensors(mgmt()); |
| BrooklynTaskTags.WorkflowTaskTag w1, w2, w3; |
| |
| doTestRetentionDisabled("disabled", "ignored", true, false, true); |
| doTestRetentionDisabled("1", "disabled", true, false, false); |
| doTestRetentionDisabled("1", "disabled", false, true, true); |
| w1 = doTestRetentionDisabled("1", "min(1,5s)", true, false, false); |
| |
| // only w1 should be persisted |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId())); |
| |
| // run something else within 5s, should now be persisting 2 |
| |
| w2 = doTestRetentionDisabled("1", "min(1,5s)", true, false, false); |
| |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId())); |
| |
| // wait 5s and run something, it should cause everything else to expire |
| Time.sleep(Duration.FIVE_SECONDS); |
| wp.get().expireOldWorkflows(app, null); |
| // should now be empty |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of()); |
| |
| String longWait = "10s"; |
| |
| w1 = doTestRetentionDisabled("1", "hash my-fixed-hash max(context,"+longWait+")", false, true, false); |
| Asserts.assertEquals(lastWorkflowContext.getRetentionSettings().expiryResolved, "max(1,"+Duration.of(longWait)+")"); |
| |
| w2 = doTestRetentionDisabled("disabled", "max(1,"+longWait+") hash my-fixed-hash", false, true, false); |
| Time.sleep(Duration.seconds(5)); |
| w3 = doTestRetentionDisabled("hash my-fixed-hash max(1,"+longWait+")", "context", false, true, false); |
| // should now have all 3 |
| wp.get().expireOldWorkflows(app, null); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId(), w3.getWorkflowId())); |
| |
| Time.sleep(Duration.seconds(5)); |
| // now just the last 1 (only 1 in 10s) |
| wp.get().expireOldWorkflows(app, null); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w3.getWorkflowId())); |
| |
| Time.sleep(Duration.seconds(5)); |
| // still have last 1 (even after 10s) |
| wp.get().expireOldWorkflows(app, null); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w3.getWorkflowId())); |
| |
| // run two more, that's all we should have |
| w1 = doTestRetentionDisabled("1", "hash my-fixed-hash", false, true, false); |
| w2 = doTestRetentionDisabled("1", "context", false, true, false); |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId())); |
| } |
| |
| int effNameCount = 0; |
| BrooklynTaskTags.WorkflowTaskTag doTestRetentionDisabled(Object retentionOnWorkflow, String retentionOnStep, boolean rebindBeforeRetentionChangedOnStep, boolean rebindAfterRetentionChangedOnStep, boolean shouldBeDisabled) throws Exception { |
| effNameCount++; |
| AttributeSensor<Integer> a = Sensors.newIntegerSensor("a"); |
| AttributeSensor<Integer> b = Sensors.newIntegerSensor("b"); |
| ((EntityInternal.SensorSupportInternal)app.sensors()).remove(a); |
| ((EntityInternal.SensorSupportInternal)app.sensors()).remove(b); |
| |
| Task<?> out = runStepsOnExistingApp("myWorkflow"+effNameCount, |
| MutableList.of( |
| "log wait for a in ${workflow.id} ${workflow.name}", |
| "wait ${entity.attributeWhenReady.a}", |
| "log got a", |
| "let result = ${entity.sensor.a}", |
| "workflow retention "+retentionOnStep, |
| "log retention step", |
| "wait ${entity.attributeWhenReady.b}", |
| "let result = ${result} + ${entity.sensor.b}", |
| "return ${result}"), |
| null, |
| ConfigBag.newInstance().configure((ConfigKey)WorkflowCommonConfig.RETENTION, retentionOnWorkflow)); |
| |
| BrooklynTaskTags.WorkflowTaskTag wt = BrooklynTaskTags.getWorkflowTaskTag(out, false); |
| |
| if (rebindBeforeRetentionChangedOnStep) { |
| Time.sleep(Duration.millis(500)); |
| app = rebind(); |
| switchOriginalToNewManagementContext(); |
| } |
| app.sensors().set(a, 10); |
| if (rebindAfterRetentionChangedOnStep) { |
| Time.sleep(Duration.millis(500)); |
| app = rebind(); |
| switchOriginalToNewManagementContext(); |
| } |
| app.sensors().set(b, 1); |
| |
| Maybe<WorkflowExecutionContext> wf = new WorkflowStatePersistenceViaSensors(mgmt()).getFromTag(wt); |
| |
| if (shouldBeDisabled) { |
| Time.sleep(Duration.millis(500)); |
| if (wf.isPresent()) Asserts.fail("Workflow persistence should be disabled, instead found: "+wf.get()); |
| |
| } else { |
| if (wf.isAbsent()) Asserts.fail("Workflow persistence should be enabled, instead did not find it"); |
| lastWorkflowContext = wf.get(); |
| lastInvocation = lastWorkflowContext.getTask(false).get(); |
| if (rebindAfterRetentionChangedOnStep || rebindBeforeRetentionChangedOnStep) { |
| // was interrupted, make sure the dangling handler finishes then replay |
| lastInvocation.blockUntilEnded(Duration.FIVE_SECONDS); |
| lastInvocation = Entities.submit(app, lastWorkflowContext.factory(false).createTaskReplaying(lastWorkflowContext.factory(false).makeInstructionsForReplayResuming("test", true))); |
| } |
| |
| Asserts.assertEquals(lastInvocation.getUnchecked(Duration.seconds(5)), 11); |
| } |
| return wt; |
| } |
| |
| @Test |
| public void testParseErrorStatusAndRetention() throws Exception { |
| app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| Supplier<WorkflowStatePersistenceViaSensors> wp = () -> new WorkflowStatePersistenceViaSensors(mgmt()); |
| |
| // non-parseable step fails at add time, and is not persisted |
| try { |
| WorkflowEffector eff = addEffector("e1", MutableList.of( |
| "non-parseable-step"), |
| null, |
| ConfigBag.newInstance().configure((ConfigKey) WorkflowCommonConfig.RETENTION, 1)); |
| Asserts.shouldHaveFailedPreviously("Instead had: "+eff); |
| } catch (Exception e) { |
| Asserts.expectedFailureContains(e, "non-parseable-step"); |
| } |
| // nothing persisted |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of()); |
| |
| // with condition, it does not parse until actual call |
| addEffector("e1", MutableList.of( |
| "non-parseable-step"), |
| null, |
| ConfigBag.newInstance() |
| .configure((ConfigKey) WorkflowCommonConfig.CONDITION, "any: []") |
| .configure((ConfigKey) WorkflowCommonConfig.RETENTION, 1) |
| ); |
| // but fails on invocation, before task created |
| try { |
| Task<?> task = app.invoke(app.getEntityType().getEffectorByName("e1").get(), null); |
| Asserts.shouldHaveFailedPreviously("Instead had: " + task.getStatusDetail(true)); |
| } catch (Exception e) { |
| Asserts.expectedFailureContains(e, "non-parseable-step"); |
| } |
| // nothing persisted |
| Asserts.assertEquals(wp.get().getWorkflows(app).keySet(), MutableSet.of()); |
| |
| // invalid replayable step also fails before persisted |
| try { |
| WorkflowBasicTest.runWorkflow(app, |
| Strings.lines( |
| "steps:", |
| "- workflow replayable unknown-replayable-mode"), |
| "e2"); |
| } catch (Exception e) { |
| Asserts.expectedFailureContains(e, "unknown-replayable-mode"); |
| } |
| Asserts.assertSize(wp.get().getWorkflows(app).keySet(), 0); |
| |
| // (above is the result after fixing bug where some parses happened in the task, and again in the error handler, |
| // and didn't properly handle errors; now all parse errors should prevent workflow creation; |
| // there might be other edge cases where workflows remain in "staged" state, but they should be few if any, |
| // and worst case they can be manually deleted) |
| } |
| |
| @Test |
| public void testErrorInSubWorkflowCaughtUpdatesContextAndStep() throws Exception { |
| app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| WorkflowExecutionContext run = WorkflowBasicTest.runWorkflow(app, Strings.lines( |
| "steps:", |
| "- log 1", |
| "- type: workflow", |
| " steps:", |
| " - log 2-1", |
| " - step: fail message 2-2", |
| " on-error:", |
| " - log 2-2-error", |
| " - fail message 2-2-done", |
| " - log 2-3", |
| " on-error: ", |
| " - return 2-done", |
| "- log 3" |
| ), "test-error-in-subworkflow"); |
| Asserts.assertEquals(run.getTask(true).get().getUnchecked(), "2-done"); |
| |
| WorkflowExecutionContext.OldStepRecord step2 = run.oldStepInfo.get(1); |
| Asserts.assertNotNull(step2); |
| Asserts.assertNotNull(step2.context); |
| Asserts.assertNull(step2.context.getError()); // should be null because handled |
| Asserts.assertNull(step2.context.errorHandlerTaskId); // should be null because not treated as a step handler, but handler for the workflow - step2sub.errorHandlerTaskId |
| |
| BrooklynTaskTags.WorkflowTaskTag step2subTag = Iterables.getOnlyElement(step2.context.getSubWorkflows()); |
| |
| WorkflowExecutionContext step2sub = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).get(step2subTag.getWorkflowId()); |
| Asserts.assertEquals(step2sub.getStatus(), WorkflowExecutionContext.WorkflowStatus.SUCCESS); |
| Asserts.assertNotNull(step2sub.errorHandlerTaskId); |
| |
| WorkflowExecutionContext.OldStepRecord step22 = step2sub.oldStepInfo.get(1); |
| Asserts.assertNotNull(step22); |
| |
| Asserts.assertNotNull(step22); |
| Asserts.assertNotNull(step22.context); |
| Asserts.assertNotNull(step22.context.getError()); // not null because not handled here |
| Asserts.assertNotNull(step22.context.errorHandlerTaskId); |
| } |
| |
| @Test |
| public void testErrorInSubWorkflowUncaughtUpdatesContextAndStep() throws Exception { |
| app = mgmt().getEntityManager().createEntity(EntitySpec.create(BasicApplication.class)); |
| WorkflowExecutionContext run = WorkflowBasicTest.runWorkflow(app, Strings.lines( |
| "steps:", |
| "- log 1", |
| "- type: workflow", |
| " steps:", |
| " - log 2-1", |
| " - step: fail message 2-2", |
| " on-error:", |
| " - log 2-2-error", |
| " - fail message 2-2-done", |
| " - log 2-3", |
| "- log 3" |
| ), "test-error-in-subworkflow"); |
| run.getTask(true).get().blockUntilEnded(); |
| Asserts.assertEquals(run.getStatus(), WorkflowExecutionContext.WorkflowStatus.ERROR); |
| |
| WorkflowExecutionContext.OldStepRecord step2 = run.oldStepInfo.get(1); |
| BrooklynTaskTags.WorkflowTaskTag step2subTag = Iterables.getOnlyElement(step2.context.getSubWorkflows()); |
| |
| WorkflowExecutionContext step2sub = new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(app).get(step2subTag.getWorkflowId()); |
| Asserts.assertEquals(step2sub.getStatus(), WorkflowExecutionContext.WorkflowStatus.ERROR); |
| |
| WorkflowExecutionContext.OldStepRecord step22 = step2sub.oldStepInfo.get(1); |
| Asserts.assertNotNull(step22); |
| Asserts.assertNotNull(step22.context); |
| Asserts.assertNotNull(step22.context.getError()); |
| Asserts.assertNotNull(step22.context.errorHandlerTaskId); |
| } |
| } |
