expire appropriate softly-kept in memory completed workflows, using system default
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlTest.java
index b7ba499..a4dc8d3 100644
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlTest.java
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlTest.java
@@ -362,7 +362,7 @@
defs.forEach(def -> Asserts.assertThat(def, d -> !(d instanceof WorkflowStepDefinition)));
} else {
EntityAsserts.assertAttributeEqualsContinually(entity, MY_WORKFLOW_SENSOR, null);
- Asserts.assertThat(new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(entity).values(), Collection::isEmpty);
+ Asserts.assertThat(new WorkflowStatePersistenceViaSensors(mgmt()).getWorkflows(entity, false).values(), Collection::isEmpty);
}
return entity;
}
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowRetentionAndExpiration.java b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowRetentionAndExpiration.java
index c6c8a16..4c2c667 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowRetentionAndExpiration.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowRetentionAndExpiration.java
@@ -106,7 +106,7 @@
static ThreadLocal<Set<String>> INIT_REENTRANT = new ThreadLocal<Set<String>>();
- static Map<String, WorkflowExecutionContext> recomputeExpiration(Map<String, WorkflowExecutionContext> v, WorkflowExecutionContext optionalContext) {
+ static Map<String, WorkflowExecutionContext> recomputeExpiration(Map<String, WorkflowExecutionContext> v, WorkflowExecutionContext optionalContext, boolean useSoftlyKeptExpiry) {
Set<String> workflowHashesToUpdate = optionalContext!=null ? MutableSet.of(Strings.firstNonBlank(optionalContext.getRetentionHash(), "empty-expiry-hash")) //should always be set
: v.values().stream().map(WorkflowExecutionContext::getRetentionHash).collect(Collectors.toSet());
@@ -121,7 +121,9 @@
Optional<WorkflowExecutionContext> existingRetentionExpiry = finishedTwins.stream().filter(w -> w.getRetentionSettings().expiry != null).findAny();
WorkflowRetentionParser.WorkflowRetentionFilter expiry;
- if (existingRetentionExpiry.isPresent()) {
+ if (useSoftlyKeptExpiry) {
+ expiry = WorkflowRetentionParser.newDefaultSoftFilter().init(finishedTwins.iterator().next());
+ } else if (existingRetentionExpiry.isPresent()) {
// log if expiry fn differs for the same hash
// (but note if it refers to parents, invocations from different places could result in different expiry functions)
if (optionalContext!=null && optionalContext.getRetentionHash().equals(k)) {
@@ -144,7 +146,7 @@
toRemove.removeAll(retainedFinishedTwins);
toRemove.forEach(w -> {
log.debug("Expiring old workflow " + w + " as there are "+retainedFinishedTwins.size()+" more recent ones also completed");
- deleteWorkflowFromMap(v, w, true);
+ deleteWorkflowFromMap(v, w, true, false);
});
}
});
@@ -152,9 +154,9 @@
return v;
}
- static boolean deleteWorkflowFromMap(Map<String, WorkflowExecutionContext> v, WorkflowExecutionContext w, boolean andAllReplayTasks) {
+ static boolean deleteWorkflowFromMap(Map<String, WorkflowExecutionContext> v, WorkflowExecutionContext w, boolean andAllReplayTasks, boolean andSoftlyKept) {
boolean removed = v.remove(w.getWorkflowId()) != null;
- removed |= WorkflowStateActiveInMemory.get(w.getManagementContext()).deleteWorkflow(w);
+ if (andSoftlyKept) removed = WorkflowStateActiveInMemory.get(w.getManagementContext()).deleteWorkflow(w) || removed;
if (andAllReplayTasks) {
BasicExecutionManager em = ((BasicExecutionManager) w.getManagementContext().getExecutionManager());
w.getReplays().forEach(wr -> {
@@ -190,6 +192,6 @@
public static void expireOldWorkflows(Entity entity) {
- new WorkflowStatePersistenceViaSensors(((EntityInternal)entity).getManagementContext()).updateMap(entity, true, true, null);
+ new WorkflowStatePersistenceViaSensors(((EntityInternal)entity).getManagementContext()).updateMaps(entity, true, true, null, null);
}
}
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStateActiveInMemory.java b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStateActiveInMemory.java
index 59f5780..88fe231 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStateActiveInMemory.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStateActiveInMemory.java
@@ -159,4 +159,11 @@
}
return result;
}
+
+ public void recomputeExpiration(Entity entity) {
+ withSoftlyKeptForEntity(entity.getId(), false, wfm -> {
+ WorkflowRetentionAndExpiration.recomputeExpiration(wfm.asMap(), null, true);
+ return null;
+ });
+ }
}
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStatePersistenceViaSensors.java b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStatePersistenceViaSensors.java
index 82ef7b0..97c37fc 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStatePersistenceViaSensors.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/store/WorkflowStatePersistenceViaSensors.java
@@ -131,7 +131,7 @@
boolean interrupted = Thread.interrupted();
boolean doExpiry = WorkflowRetentionAndExpiration.isExpirationCheckNeeded(entity);
try {
- return updateMap(entity, doExpiry, true, context==null ? null : v -> v.put(context.getWorkflowId(), context));
+ return updateMaps(entity, doExpiry, true, context==null ? null : v -> v.put(context.getWorkflowId(), context), null);
} finally {
if (interrupted) Thread.currentThread().interrupt();
@@ -142,10 +142,10 @@
if (w.getStatus()==null || w.getStatus().expirable || w.getStatus()== WorkflowExecutionContext.WorkflowStatus.STAGED) {
log.debug("Explicit request to delete workflow "+w);
AtomicBoolean result = new AtomicBoolean(false);
- updateMap(w.getEntity(), false, true, map -> {
- boolean removed = WorkflowRetentionAndExpiration.deleteWorkflowFromMap(map, w, true);
+ updateMaps(w.getEntity(), false, true, map -> {
+ boolean removed = WorkflowRetentionAndExpiration.deleteWorkflowFromMap(map, w, true, true);
if (removed) result.set(true);
- });
+ }, w);
return result.get();
} else {
log.warn("Explicit request to delete non-expirable workflow "+w+"; ignoring");
@@ -153,13 +153,26 @@
}
}
+ int updateMaps(Entity entity, boolean doExpiry, boolean persist, Consumer<Map<String,WorkflowExecutionContext>> action, WorkflowExecutionContext contextToRemoveFromSoftMemory) {
+ int result = updateMap(entity, doExpiry, persist, action);
+
+ // and update softly kept
+ WorkflowStateActiveInMemory activeInMemory = WorkflowStateActiveInMemory.get(mgmt);
+ if (contextToRemoveFromSoftMemory!=null) {
+ activeInMemory.deleteWorkflow(contextToRemoveFromSoftMemory);
+ }
+ if (doExpiry) activeInMemory.recomputeExpiration(entity);
+
+ return result;
+ }
+
int updateMap(Entity entity, boolean doExpiry, boolean persist, Consumer<Map<String,WorkflowExecutionContext>> action) {
AtomicInteger delta = new AtomicInteger(0);
entity.sensors().modify(INTERNAL_WORKFLOWS, vo -> {
Map<String, WorkflowExecutionContext> v = MutableMap.copyOf(vo);
delta.set(-v.size());
if (action!=null) action.accept(v);
- if (doExpiry) v = WorkflowRetentionAndExpiration.recomputeExpiration(v, null);
+ if (doExpiry) v = WorkflowRetentionAndExpiration.recomputeExpiration(v, null, false);
delta.getAndAdd(v.size());
return Maybe.of(v);
});
@@ -168,7 +181,10 @@
}
public Map<String,WorkflowExecutionContext> getWorkflows(Entity entity) {
- MutableMap<String, WorkflowExecutionContext> result = WorkflowStateActiveInMemory.get(mgmt).getWorkflowsCopy(entity, true);
+ return getWorkflows(entity, true);
+ }
+ public Map<String,WorkflowExecutionContext> getWorkflows(Entity entity, boolean includeSoftlyKeptCompleted) {
+ MutableMap<String, WorkflowExecutionContext> result = WorkflowStateActiveInMemory.get(mgmt).getWorkflowsCopy(entity, includeSoftlyKeptCompleted);
result.add(entity.sensors().get(INTERNAL_WORKFLOWS));
return result;
}
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/utils/WorkflowRetentionParser.java b/core/src/main/java/org/apache/brooklyn/core/workflow/utils/WorkflowRetentionParser.java
index d248845..a5dbe86 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/utils/WorkflowRetentionParser.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/utils/WorkflowRetentionParser.java
@@ -227,6 +227,9 @@
public static WorkflowRetentionFilter newDefaultFilter() {
return new KeepParent();
}
+ public static WorkflowRetentionFilter newDefaultSoftFilter() {
+ return new KeepSystem();
+ }
static class KeepParent extends KeepDelegate {
@Override
public WorkflowRetentionFilter findDelegate(WorkflowExecutionContext workflow) {
diff --git a/core/src/test/java/org/apache/brooklyn/core/workflow/WorkflowPersistReplayErrorsTest.java b/core/src/test/java/org/apache/brooklyn/core/workflow/WorkflowPersistReplayErrorsTest.java
index adc6fd2..f90a0a2 100644
--- a/core/src/test/java/org/apache/brooklyn/core/workflow/WorkflowPersistReplayErrorsTest.java
+++ b/core/src/test/java/org/apache/brooklyn/core/workflow/WorkflowPersistReplayErrorsTest.java
@@ -909,29 +909,40 @@
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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of(w1.getWorkflowId()));
+
+ // softly in-memory should also have both
+ Asserts.assertSize(wp.get().getWorkflows(app, true).keySet(), 2);
+
+ // do it 3 more times, we should have 1 persisted, but 3 in memory (not 4)
+ w1 = doTestRetentionDisabled(2, "hash my-fixed-hash min(1,context)", false, false, false);
+ w1 = doTestRetentionDisabled(2, "hash my-fixed-hash min(1,context)", false, false, false);
+ w1 = doTestRetentionDisabled(2, "hash my-fixed-hash min(1,context)", false, false, false);
+
+ Asserts.assertSize(wp.get().getWorkflows(app, false).keySet(), 1);
+ Asserts.assertSize(wp.get().getWorkflows(app, true).keySet(), 3);
// 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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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);
@@ -940,7 +951,7 @@
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
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId(), w3.getWorkflowId(), w4.getWorkflowId())); // should replace the one above
}
@Test(groups="Integration") // very slow
@@ -957,19 +968,19 @@
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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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());
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of());
String longWait = "10s";
@@ -981,22 +992,22 @@
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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).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()));
+ Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId()));
}
int effNameCount = 0;
