You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2024/04/01 15:38:16 UTC
(brooklyn-server) branch master updated (3c977b9b5c -> 057d8f8bc0)
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
from 3c977b9b5c suppress handled error message in processing xpath
new 931caf0bf0 record the workflow resolution context better
new 51ad80bf69 allow keeping soft references to previous workflows
new fb7c01507a expire appropriate softly-kept in memory completed workflows, using system default
new 16f868a9b9 allow soft retention to be specified, globally or per workflow/step
new 9ef5dcc7a4 pause most tasks submitted during entity startup, until all entities are managed
new 057d8f8bc0 allow from system if possible where bundle not known
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../brooklyn/api/mgmt/ManagementContext.java | 9 +-
.../camp/brooklyn/SshCommandSensorYamlTest.java | 19 +-
.../camp/brooklyn/WorkflowYamlRebindTest.java | 18 +-
.../brooklyn/camp/brooklyn/WorkflowYamlTest.java | 2 +-
.../mgmt/internal/EntityManagementSupport.java | 34 ++-
.../core/mgmt/internal/LocalManagementContext.java | 3 +-
.../core/workflow/WorkflowExecutionContext.java | 7 +-
.../workflow/WorkflowExpressionResolution.java | 247 ++++++++++++---------
.../store/WorkflowRetentionAndExpiration.java | 191 ++++++++++++----
.../store/WorkflowStateActiveInMemory.java | 125 ++++++-----
.../store/WorkflowStatePersistenceViaSensors.java | 98 ++++----
.../workflow/utils/WorkflowRetentionParser.java | 124 ++++++++---
.../brooklyn/util/core/ClassLoaderUtils.java | 6 +-
.../util/core/task/BasicExecutionContext.java | 55 ++++-
.../brooklyn/util/core/text/TemplateProcessor.java | 6 +-
.../brooklyn/core/mgmt/ha/HotStandbyTest.java | 2 +-
.../workflow/WorkflowPersistReplayErrorsTest.java | 61 +++--
.../brooklyn/util/core/ClassLoaderUtilsTest.java | 8 +-
.../system_service/SystemServiceEnricher.java | 3 +-
.../SoftwareProcessRebindNotRunningEntityTest.java | 8 +-
.../util/collections/ThreadLocalStack.java | 23 +-
21 files changed, 683 insertions(+), 366 deletions(-)
(brooklyn-server) 06/06: allow from system if possible where bundle not known
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit 057d8f8bc0010551a1e4c0e4dc05546063374084
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Mon Apr 1 16:19:37 2024 +0100
allow from system if possible where bundle not known
for use if switching to a pojo environment
---
.../main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java | 6 +++++-
.../java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java | 8 ++++----
2 files changed, 9 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java b/core/src/main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java
index fc7e9bffe2..ce127c90bd 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java
@@ -380,7 +380,11 @@ public class ClassLoaderUtils {
.reduce((v1, v2) -> v2));
}
if (bundle.isAbsent()) {
- throw new IllegalStateException("Bundle " + toBundleString(symbolicName, version)+ " not found to load.");
+ // fall back to loading from system classpath, if running as pojo
+ Maybe<T> result = dispatcher.tryLoadFrom(classLoader, name);
+ if (result.isAbsent()) {
+ throw new IllegalStateException("Bundle " + toBundleString(symbolicName, version) + " not found to load.");
+ }
}
}
return dispatcher.tryLoadFrom(bundle.get(), name);
diff --git a/core/src/test/java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java b/core/src/test/java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java
index efa0aea290..7b6df6d797 100644
--- a/core/src/test/java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java
+++ b/core/src/test/java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java
@@ -308,23 +308,23 @@ public class ClassLoaderUtilsTest {
new ClassLoaderUtils(this, mgmt).loadClass(
"org.apache.brooklyn.api",
"100.100.100-alpha-version_wth.tags",
- Entity.class.getName());
+ "org.apache.brooklyn.NonExistentClass"); // actual classes may be found on the classpath
Asserts.shouldHaveFailedPreviously();
} catch (Exception e) {
IllegalStateException nested = Exceptions.getFirstThrowableOfType(e, IllegalStateException.class);
assertNotNull(nested);
- Asserts.expectedFailureContains(nested, "not found to load");
+ Asserts.expectedFailureContains(nested, "not found to load", "org.apache.brooklyn.api", "100.100.100");
}
try {
new ClassLoaderUtils(this, mgmt).loadClass(
"org.apache.brooklyn.api",
"100.100.100-SNAPSHOT",
- Entity.class.getName());
+ "org.apache.brooklyn.NonExistentClass"); // actual classes may be found on the classpath
Asserts.shouldHaveFailedPreviously();
} catch (Exception e) {
IllegalStateException nested = Exceptions.getFirstThrowableOfType(e, IllegalStateException.class);
assertNotNull(nested);
- Asserts.expectedFailureContains(nested, "not found to load");
+ Asserts.expectedFailureContains(nested, "not found to load", "org.apache.brooklyn.api", "100.100.100");
}
}
(brooklyn-server) 02/06: allow keeping soft references to previous workflows
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit 51ad80bf693adc11085d219842bf529ad04257b9
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Fri Mar 29 23:54:50 2024 +0000
allow keeping soft references to previous workflows
means we can set 'retention: disabled' for persistence purposes but still - usually - have access to them in memory.
does not yet apply any limits to that cache, but thinking is to follow system default for now,
and perhaps to allow configuration in future.
---
.../camp/brooklyn/WorkflowYamlRebindTest.java | 18 ++--
.../store/WorkflowRetentionAndExpiration.java | 7 +-
.../store/WorkflowStateActiveInMemory.java | 106 ++++++++++++---------
.../store/WorkflowStatePersistenceViaSensors.java | 13 ++-
.../workflow/utils/WorkflowRetentionParser.java | 34 +++++--
.../workflow/WorkflowPersistReplayErrorsTest.java | 2 +-
6 files changed, 111 insertions(+), 69 deletions(-)
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlRebindTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlRebindTest.java
index b4b1145185..697428ccec 100644
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlRebindTest.java
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/WorkflowYamlRebindTest.java
@@ -18,7 +18,10 @@
*/
package org.apache.brooklyn.camp.brooklyn;
-import com.google.common.base.Stopwatch;
+import java.io.File;
+import java.util.Map;
+import java.util.Set;
+
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
@@ -28,20 +31,20 @@ import org.apache.brooklyn.api.entity.EntitySpec;
import org.apache.brooklyn.api.location.LocationSpec;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.objs.EntityAdjunct;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.api.sensor.Feed;
import org.apache.brooklyn.camp.brooklyn.spi.dsl.methods.BrooklynDslCommon;
import org.apache.brooklyn.core.config.ConfigKeys;
-import org.apache.brooklyn.core.entity.*;
+import org.apache.brooklyn.core.entity.Dumper;
+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.LocalManagementContext;
-import org.apache.brooklyn.core.mgmt.rebind.RebindOptions;
import org.apache.brooklyn.core.sensor.Sensors;
import org.apache.brooklyn.core.test.entity.TestApplication;
import org.apache.brooklyn.core.test.entity.TestEntity;
import org.apache.brooklyn.core.workflow.WorkflowBasicTest;
import org.apache.brooklyn.core.workflow.WorkflowEffector;
-import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
import org.apache.brooklyn.entity.software.base.EmptySoftwareProcess;
import org.apache.brooklyn.entity.stock.BasicApplication;
import org.apache.brooklyn.entity.stock.BasicEntity;
@@ -59,11 +62,6 @@ import org.slf4j.LoggerFactory;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.io.File;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.Predicate;
-
public class WorkflowYamlRebindTest extends AbstractYamlRebindTest {
private static final Logger log = LoggerFactory.getLogger(WorkflowYamlRebindTest.class);
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 cc0adc5e9b..c6c8a16d73 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
@@ -50,9 +50,14 @@ public class WorkflowRetentionAndExpiration {
public static class WorkflowRetentionSettings {
public Boolean disabled;
public String hash;
+
public String expiry;
public String expiryResolved;
+ // soft expiry refers to what is kept in memory
+ public String softExpiry;
+ public String softExpiryResolved;
+
@JsonIgnore
private transient WorkflowRetentionParser.WorkflowRetentionFilter expiryFn;
@@ -163,7 +168,7 @@ public class WorkflowRetentionAndExpiration {
private static boolean isExpirable(WorkflowExecutionContext c) {
if (c.getStatus() == null || !c.getStatus().expirable) return false;
if (c.getParent()!=null) {
- // fow now, don't expire children workflows unless parents are also expirable
+ // XXX for size reasons, should skip this - fow now, don't expire children workflows unless parents are also expirable
if (!isExpirable(c.getParent())) return false;
// we could weaken this if we have lots of children workflows, but that is more work; left as an enhancement
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 d27a161a16..59f5780a87 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
@@ -20,13 +20,17 @@ package org.apache.brooklyn.core.workflow.store;
import java.util.Map;
import java.util.Set;
+import java.util.function.Function;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ManagementContext;
import org.apache.brooklyn.config.ConfigKey;
import org.apache.brooklyn.core.config.ConfigKeys;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
+import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.collections.MutableSet;
import org.slf4j.Logger;
@@ -55,7 +59,10 @@ public class WorkflowStateActiveInMemory {
}
private final ManagementContext mgmt;
- final Map<String,Map<String,WorkflowExecutionContext>> data = MutableMap.of();
+ // active workflows by entity then workflow id
+ final Map<String,Map<String,WorkflowExecutionContext>> active = MutableMap.of();
+ // cached remembered workflows by entity then workflow id
+ final Map<String, Cache<String, WorkflowExecutionContext>> completedSoftlyKept = MutableMap.of();
// created and managed by mgmt context scratchpad
protected WorkflowStateActiveInMemory(ManagementContext mgmt) {
@@ -67,40 +74,25 @@ public class WorkflowStateActiveInMemory {
public void expireAbsentEntities() {
lastInMemClear = System.currentTimeMillis();
Set<String> copy;
- synchronized (data) {
- copy = MutableSet.copyOf(data.keySet());
- }
+ synchronized (active) { copy = MutableSet.copyOf(active.keySet()); }
+ synchronized (completedSoftlyKept) { copy.addAll(completedSoftlyKept.keySet()); }
+
copy.forEach(entityId -> {
if (mgmt.getEntityManager().getEntity(entityId) == null) {
- data.remove(entityId);
+ synchronized (active) { active.remove(entityId); }
+ synchronized (completedSoftlyKept) { completedSoftlyKept.remove(entityId); }
}
});
}
public void checkpoint(WorkflowExecutionContext context) {
- // keep active workflows in memory, even if disabled
- Map<String, WorkflowExecutionContext> entityActiveWorkflows = getForWorkflowIdWithLockButResultNeedsSynch(context.getEntity().getId());
if (context.getStatus().expirable) {
- if (entityActiveWorkflows!=null) {
- synchronized (entityActiveWorkflows) {
- entityActiveWorkflows.remove(context.getWorkflowId());
- }
- }
+ withActiveForEntity(context.getEntity().getId(), false, wfm -> wfm.remove(context.getWorkflowId()));
+ withSoftlyKeptForEntity(context.getEntity().getId(), true, wfm -> { wfm.put(context.getWorkflowId(), context); return null; });
} else {
- if (entityActiveWorkflows==null) {
- synchronized (data) {
- entityActiveWorkflows = data.get(context.getEntity().getId());
- if (entityActiveWorkflows==null) {
- entityActiveWorkflows = MutableMap.of();
- data.put(context.getEntity().getId(), entityActiveWorkflows);
- }
- }
- }
- synchronized (entityActiveWorkflows) {
- entityActiveWorkflows.put(context.getWorkflowId(), context);
- }
+ // keep active workflows in memory, even if disabled
+ withActiveForEntity(context.getEntity().getId(), true, wfm -> wfm.put(context.getWorkflowId(), context));
}
-
if (lastInMemClear + GLOBAL_UPDATE_FREQUENCY < System.currentTimeMillis()) {
// poor man's cleanup, every minute, but good enough
expireAbsentEntities();
@@ -112,37 +104,59 @@ public class WorkflowStateActiveInMemory {
return getWorkflowsCopy(entity);
}
public MutableMap<String,WorkflowExecutionContext> getWorkflowsCopy(Entity entity) {
- Map<String, WorkflowExecutionContext> entityActiveWorkflows = getForWorkflowIdWithLockButResultNeedsSynch(entity.getId());
- if (entityActiveWorkflows == null) return MutableMap.of();
- synchronized (entityActiveWorkflows) {
- return MutableMap.copyOf(entityActiveWorkflows);
- }
+ return getWorkflowsCopy(entity, true);
+ }
+ public MutableMap<String,WorkflowExecutionContext> getWorkflowsCopy(Entity entity, boolean includeSoftlyKeptCompleted) {
+ MutableMap<String,WorkflowExecutionContext> result = MutableMap.of();
+ withActiveForEntity(entity.getId(), false, wfm -> { result.putAll(wfm); return null; });
+ if (includeSoftlyKeptCompleted) withSoftlyKeptForEntity(entity.getId(), false, wfm -> { result.putAll(wfm.asMap()); return null; });
+ return result;
}
boolean deleteWorkflow(WorkflowExecutionContext context) {
- Map<String, WorkflowExecutionContext> entityActiveWorkflows = getForWorkflowIdWithLockButResultNeedsSynch(context.getEntity().getId());
- if (entityActiveWorkflows!=null) {
- synchronized (entityActiveWorkflows) {
- return entityActiveWorkflows.remove(context.getWorkflowId()) != null;
- }
+ boolean result = false;
+ result = Boolean.TRUE.equals(withActiveForEntity(context.getEntity().getId(), false, wfm -> wfm.remove(context.getWorkflowId())!=null)) || result;
+ result = Boolean.TRUE.equals(withSoftlyKeptForEntity(context.getEntity().getId(), false, wfm -> {
+ WorkflowExecutionContext soft = wfm.getIfPresent(context.getWorkflowId());
+ wfm.invalidate(context.getWorkflowId());
+ return (soft!=null);
+ })) || result;
+
+ return result;
+ }
+
+ private <T> T withActiveForEntity(String entityId, boolean upsert, Function<Map<String, WorkflowExecutionContext>,T> fn) {
+ if (entityId==null) return null;
+ Map<String, WorkflowExecutionContext> result;
+ synchronized (active) {
+ result = active.computeIfAbsent(entityId, _key -> upsert ? MutableMap.of() : null);
+ }
+ if (result==null) return null;
+ synchronized (result) {
+ return fn.apply(result);
}
- return false;
}
- // note: callers should subsequently sync on the returned map
- private Map<String, WorkflowExecutionContext> getForWorkflowIdWithLockButResultNeedsSynch(String entityId) {
- synchronized (data) {
- return data.get(entityId);
+ private <T> T withSoftlyKeptForEntity(String entityId, boolean upsert, Function<Cache<String, WorkflowExecutionContext>,T> fn) {
+ if (entityId==null) return null;
+ Cache<String, WorkflowExecutionContext> result;
+ synchronized (completedSoftlyKept) {
+ result = completedSoftlyKept.computeIfAbsent(entityId, _key -> upsert ? CacheBuilder.newBuilder().softValues().build() : null);
+ }
+ if (result==null) return null;
+ synchronized (result) {
+ return fn.apply(result);
}
}
public WorkflowExecutionContext getFromTag(BrooklynTaskTags.WorkflowTaskTag tag) {
- Map<String, WorkflowExecutionContext> activeForEntity = getForWorkflowIdWithLockButResultNeedsSynch(tag.getEntityId());
- if (activeForEntity!=null) {
- synchronized (activeForEntity) {
- return activeForEntity.get(tag.getWorkflowId());
- }
+ return getFromTag(tag, true);
+ }
+ public WorkflowExecutionContext getFromTag(BrooklynTaskTags.WorkflowTaskTag tag, boolean includeCompletedSoftlyKept) {
+ WorkflowExecutionContext result = withActiveForEntity(tag.getEntityId(), false, wfm -> wfm.get(tag.getWorkflowId()));
+ if (includeCompletedSoftlyKept && result==null) {
+ result = withSoftlyKeptForEntity(tag.getEntityId(), false, wfm -> wfm.getIfPresent(tag.getWorkflowId()));
}
- return null;
+ return result;
}
}
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 63170c539b..82ef7b0c30 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
@@ -94,7 +94,7 @@ public class WorkflowStatePersistenceViaSensors {
WorkflowStateActiveInMemory.get(context.getManagementContext()).checkpoint(context);
if (Boolean.TRUE.equals(context.getRetentionSettings().disabled)) {
- if (getFromTag(BrooklynTaskTags.tagForWorkflow(context), false)!=null) {
+ if (getFromTag(BrooklynTaskTags.tagForWorkflow(context), false, false)!=null) {
// need to clear
updateMap(entity, false, true, v -> v.remove(context.getWorkflowId(), context));
}
@@ -168,7 +168,7 @@ public class WorkflowStatePersistenceViaSensors {
}
public Map<String,WorkflowExecutionContext> getWorkflows(Entity entity) {
- MutableMap<String, WorkflowExecutionContext> result = WorkflowStateActiveInMemory.get(mgmt).getWorkflowsCopy(entity);
+ MutableMap<String, WorkflowExecutionContext> result = WorkflowStateActiveInMemory.get(mgmt).getWorkflowsCopy(entity, true);
result.add(entity.sensors().get(INTERNAL_WORKFLOWS));
return result;
}
@@ -185,17 +185,20 @@ public class WorkflowStatePersistenceViaSensors {
}
public Maybe<WorkflowExecutionContext> getFromTag(BrooklynTaskTags.WorkflowTaskTag tag) {
- return getFromTag(tag, true);
+ return getFromTag(tag, true, true);
+ }
+ public Maybe<WorkflowExecutionContext> getFromTag(BrooklynTaskTags.WorkflowTaskTag tag, boolean includeSoftlyKeptInMemory) {
+ return getFromTag(tag, true, includeSoftlyKeptInMemory);
}
- private Maybe<WorkflowExecutionContext> getFromTag(BrooklynTaskTags.WorkflowTaskTag tag, boolean allowInMemory) {
+ private Maybe<WorkflowExecutionContext> getFromTag(BrooklynTaskTags.WorkflowTaskTag tag, boolean allowActiveInMemory, boolean allowActiveAndSoftlyKeptInMemory) {
Entity targetEntity = mgmt.lookup(tag.getEntityId(), Entity.class);
if (targetEntity==null) {
return Maybe.absent("Entity "+tag.getWorkflowId()+" not found");
} else {
WorkflowExecutionContext w = null;
- if (allowInMemory) w = WorkflowStateActiveInMemory.get(mgmt).getFromTag(tag);
+ if (allowActiveInMemory || allowActiveAndSoftlyKeptInMemory) w = WorkflowStateActiveInMemory.get(mgmt).getFromTag(tag, allowActiveAndSoftlyKeptInMemory);
if (w==null) {
w = new WorkflowStatePersistenceViaSensors(mgmt).getWorkflows(targetEntity).get(tag.getWorkflowId());
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 18ea10d7b0..d2488457bc 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
@@ -55,6 +55,8 @@ the semantics of `min` and `max` are
* `max` means completed workflow instances must be retained if they meet any of the constraints implied by the `<value>` arguments, i.e. `max(2, 3, 1h, 2h)` means to keep the 3 most recent instances irrespective of when they run, and to keep all instances for up to two hours
also allows a `hash <value>` to be set at the start or the end
+
+also allows `hard` at start or end, or `soft [limit]` at end
*/
public static WorkflowRetentionSettings parse(String retentionExpression, @Nullable WorkflowExecutionContext context) {
@@ -63,20 +65,40 @@ also allows a `hash <value>` to be set at the start or the end
if (Strings.isBlank(retentionExpression)) return result;
retentionExpression = retentionExpression.trim().toLowerCase();
- if (retentionExpression.equals("disabled")) {
- result.disabled = true;
-
- } else {
-
+ do {
if (retentionExpression.startsWith("hash ")) {
+ if (result.hash != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hash' in retention expression");
retentionExpression = Strings.removeFromStart(retentionExpression, "hash").trim();
result.hash = Strings.getFirstWord(retentionExpression);
retentionExpression = retentionExpression.substring(result.hash.length()).trim();
- } else if (retentionExpression.contains(" hash ")) {
+ continue;
+ }
+ if (retentionExpression.startsWith("hard ")) {
+ if (result.softExpiry != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hard' or 'soft' in retention expression");
+ retentionExpression = Strings.removeFromStart(retentionExpression, "hard").trim();
+ result.softExpiry = "0";
+ continue;
+ }
+
+ // TODO soft/hard keyword; take whichever occurs last
+ if (retentionExpression.contains(" hash ")) {
+ if (result.hash != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hash' in retention expression");
int i = retentionExpression.indexOf(" hash ");
result.hash = Strings.removeFromStart(retentionExpression.substring(i).trim(), "hash").trim();
retentionExpression = retentionExpression.substring(0, i).trim();
+ continue;
}
+ break;
+ } while (false);
+
+ if (retentionExpression.equals("disabled")) {
+ result.disabled = true;
+
+ } else {
+
if (Strings.isNonBlank(result.hash) && context!=null) {
result.hash = context.resolve(WorkflowExpressionResolution.WorkflowExpressionStage.STEP_RUNNING, result.hash, String.class);
}
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 32a0328970..adc6fd2702 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
@@ -1036,7 +1036,7 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
}
app.sensors().set(b, 1);
- Maybe<WorkflowExecutionContext> wf = new WorkflowStatePersistenceViaSensors(mgmt()).getFromTag(wt);
+ Maybe<WorkflowExecutionContext> wf = new WorkflowStatePersistenceViaSensors(mgmt()).getFromTag(wt, false);
if (shouldBeDisabled) {
Time.sleep(Duration.millis(500));
(brooklyn-server) 04/06: allow soft retention to be specified, globally or per workflow/step
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit 16f868a9b918c46e32fb689f6abc7c6c8c9ff501
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Mon Apr 1 11:30:51 2024 +0100
allow soft retention to be specified, globally or per workflow/step
---
.../core/workflow/WorkflowExecutionContext.java | 7 +-
.../store/WorkflowRetentionAndExpiration.java | 184 ++++++++++++++++-----
.../store/WorkflowStateActiveInMemory.java | 18 +-
.../store/WorkflowStatePersistenceViaSensors.java | 71 +++-----
.../workflow/utils/WorkflowRetentionParser.java | 105 ++++++++----
.../workflow/WorkflowPersistReplayErrorsTest.java | 22 ++-
.../SoftwareProcessRebindNotRunningEntityTest.java | 8 +-
7 files changed, 275 insertions(+), 140 deletions(-)
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExecutionContext.java b/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExecutionContext.java
index 8cffdcbcc3..f026023198 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExecutionContext.java
@@ -633,17 +633,12 @@ public class WorkflowExecutionContext {
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);
+ WorkflowRetentionAndExpiration.checkpoint(getManagementContext(), this);
}
/** Get the value of the input. Supports Brooklyn DSL resolution but NOT Freemarker resolution. */
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 4c2c667d59..852c75d11f 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
@@ -18,9 +18,20 @@
*/
package org.apache.brooklyn.core.workflow.store;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude;
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.config.ConfigKey;
import org.apache.brooklyn.core.config.ConfigKeys;
@@ -28,6 +39,7 @@ import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
+import org.apache.brooklyn.core.workflow.store.WorkflowStatePersistenceViaSensors.PersistenceWithQueuedTasks;
import org.apache.brooklyn.core.workflow.utils.WorkflowRetentionParser;
import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.core.task.BasicExecutionManager;
@@ -36,15 +48,51 @@ import org.apache.brooklyn.util.text.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.util.stream.Collectors;
-
public class WorkflowRetentionAndExpiration {
private static final Logger log = LoggerFactory.getLogger(WorkflowRetentionAndExpiration.class);
public static final ConfigKey<String> WORKFLOW_RETENTION_DEFAULT = ConfigKeys.newStringConfigKey("workflow.retention.default",
- "Default retention for workflows", "3");
+ "Default retention for workflows (persisted)", "3");
+ public static final ConfigKey<String> WORKFLOW_RETENTION_DEFAULT_SOFT = ConfigKeys.newStringConfigKey("workflow.retention.default.soft",
+ "Default soft retention for workflows (in-memory)", "3");
+
+ public static void checkpoint(ManagementContext mgmt, WorkflowExecutionContext context) {
+ Entity entity = context.getEntity();
+ if (Entities.isUnmanagingOrNoLongerManaged(entity)) {
+ log.debug("Skipping persistence of "+context+" as entity is no longer active here");
+ return;
+ }
+
+ doGlobalUpdateIfNeededOnDiskAndInMemory(mgmt);
+
+ new WorkflowStatePersistenceViaSensors(mgmt).checkpoint(context, PersistenceWithQueuedTasks.WARN);
+
+ // keep active workflows in memory, even if disabled
+ WorkflowStateActiveInMemory.get(context.getManagementContext()).checkpoint(context);
+ }
+
+ static final long GLOBAL_UPDATE_FREQUENCY = 5*60*1000; // every 5m wipe out old workflows
+
+ static void doGlobalUpdateIfNeededOnDiskAndInMemory(ManagementContext mgmt) {
+ WorkflowStateActiveInMemory inMem = WorkflowStateActiveInMemory.get(mgmt);
+
+ if (inMem.lastGlobalClear + GLOBAL_UPDATE_FREQUENCY > System.currentTimeMillis()) return;
+ inMem.lastGlobalClear = System.currentTimeMillis();
+
+ AtomicInteger total = new AtomicInteger(0);
+ Collection<Entity> entities = mgmt.getEntityManager().getEntities();
+ entities.forEach(entity -> {
+ // on disk
+ int change = new WorkflowStatePersistenceViaSensors(mgmt).expireOldWorkflowsOnDisk(entity, null);
+ if (change!=0) log.debug("Global entity workflow persistence update, removed "+(-change)+" workflows from "+entity);
+ total.addAndGet(change);
+
+ // in memory
+ inMem.recomputeExpiration(entity, null);
+ });
+ if (total.get()!=0) log.debug("Global entity workflow persistence update, removed "+(-total.get())+" workflows across all "+entities.size()+" entities");
+ }
@JsonInclude(JsonInclude.Include.NON_EMPTY)
public static class WorkflowRetentionSettings {
@@ -60,35 +108,64 @@ public class WorkflowRetentionAndExpiration {
@JsonIgnore
private transient WorkflowRetentionParser.WorkflowRetentionFilter expiryFn;
+ @JsonIgnore
+ private transient WorkflowRetentionParser.WorkflowRetentionFilter softExpiryFn;
public WorkflowRetentionParser.WorkflowRetentionFilter getExpiryFn(WorkflowExecutionContext w) {
return init(w).expiryFn;
}
+ public WorkflowRetentionParser.WorkflowRetentionFilter getSoftExpiryFn(WorkflowExecutionContext w) {
+ return init(w).softExpiryFn;
+ }
public WorkflowRetentionSettings init(WorkflowExecutionContext w) {
if (w.getParent()!=null && Boolean.TRUE.equals(w.getParent().getRetentionSettings().disabled)) {
disabled = true;
- } else if (expiryFn==null) {
- expiryResolved = expiryResolved!=null ? expiryResolved : expiry;
- expiryFn = new WorkflowRetentionParser(expiryResolved).parse();
- if (w != null) {
- Set<String> set = INIT_REENTRANT.get();
- if (set==null) {
- set = MutableSet.of();
- INIT_REENTRANT.set(set);
- }
- if (!set.add(w.getWorkflowId()+":"+expiryResolved)) {
- // double-check we don't cause endless loops; see KeepContext notes
- throw new IllegalStateException("Invalid workflow retention '"+expiryResolved+"' as it refers to itself");
+ } else {
+ if (expiryFn == null) {
+ expiryResolved = expiryResolved != null ? expiryResolved : expiry;
+ expiryFn = new WorkflowRetentionParser(expiryResolved).parse();
+ if (w != null) {
+ Set<String> set = INIT_REENTRANT.get();
+ if (set == null) {
+ set = MutableSet.of();
+ INIT_REENTRANT.set(set);
+ }
+ if (!set.add(w.getWorkflowId() + ":" + expiryResolved)) {
+ // double-check we don't cause endless loops; see KeepContext notes
+ throw new IllegalStateException("Invalid workflow retention '" + expiryResolved + "' as it refers to itself");
+ }
+ try {
+ expiryFn = expiryFn.init(w);
+ } finally {
+ set.remove(w.getWorkflowId() + ":" + expiryResolved);
+ if (set.isEmpty()) INIT_REENTRANT.remove();
+ }
}
- try {
- expiryFn = expiryFn.init(w);
- } finally {
- set.remove(w.getWorkflowId()+":"+expiryResolved);
- if (set.isEmpty()) INIT_REENTRANT.remove();
+ expiryResolved = expiryFn.toString(); // remove any references to `context` that might trigger an infinite loop
+ }
+ if (softExpiryFn == null) {
+ softExpiryResolved = softExpiryResolved != null ? softExpiryResolved : softExpiry;
+ softExpiryFn = new WorkflowRetentionParser(softExpiryResolved).soft().parse();
+ if (w != null) {
+ Set<String> set = INIT_REENTRANT.get();
+ if (set == null) {
+ set = MutableSet.of();
+ INIT_REENTRANT.set(set);
+ }
+ if (!set.add(w.getWorkflowId() + ":" + softExpiryResolved)) {
+ // double-check we don't cause endless loops; see KeepContext notes
+ throw new IllegalStateException("Invalid workflow retention '" + softExpiryResolved + "' as it refers to itself");
+ }
+ try {
+ softExpiryFn = softExpiryFn.init(w);
+ } finally {
+ set.remove(w.getWorkflowId() + ":" + softExpiryResolved);
+ if (set.isEmpty()) INIT_REENTRANT.remove();
+ }
}
+ softExpiryResolved = softExpiryFn.toString(); // remove any references to `context` that might trigger an infinite loop
}
- expiryResolved = expiryFn.toString(); // remove any references to `context` that might trigger an infinite loop
}
return this;
}
@@ -101,42 +178,67 @@ public class WorkflowRetentionAndExpiration {
this.expiryFn = r2.expiryFn;
this.expiryResolved = r2.expiryResolved;
}
+ if (Strings.isNonEmpty(r2.softExpiry)) {
+ this.softExpiry = r2.softExpiry;
+ this.softExpiryFn = r2.softExpiryFn;
+ this.softExpiryResolved = r2.softExpiryResolved;
+ }
}
}
static ThreadLocal<Set<String>> INIT_REENTRANT = new ThreadLocal<Set<String>>();
- static Map<String, WorkflowExecutionContext> recomputeExpiration(Map<String, WorkflowExecutionContext> v, WorkflowExecutionContext optionalContext, boolean useSoftlyKeptExpiry) {
+ static Map<String, WorkflowExecutionContext> recomputeExpiration(Map<String, WorkflowExecutionContext> v, @Nullable 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());
workflowHashesToUpdate.forEach(k -> {
+ // if optional context supplied, perhaps only recompute for that hash
+ if (optionalContext!=null && !k.equals(optionalContext.getRetentionHash())) {
+ if (!isExpirable(optionalContext)) {
+ return;
+ } else {
+ // no-op -- if it is expirable, do a full recompute for the entity, to ensure sub-workflows are no longer retained
+ // (cross-entity subworkflows will not be cleaned up; they will get collected when another workflow runs there,
+ // or when there is a global cleanup event)
+ }
+ }
+
List<WorkflowExecutionContext> finishedTwins = v.values().stream()
.filter(c -> k.equals(c.getRetentionHash()))
.filter(c -> isExpirable(c))
- .filter(c -> !c.equals(optionalContext))
.collect(Collectors.toList());
if (finishedTwins.isEmpty()) return;
- Optional<WorkflowExecutionContext> existingRetentionExpiry = finishedTwins.stream().filter(w -> w.getRetentionSettings().expiry != null).findAny();
+ Function<WorkflowRetentionSettings,String> expiryAccessor = useSoftlyKeptExpiry ? wrs -> wrs.softExpiry : wrs -> wrs.expiry;
+ Optional<WorkflowExecutionContext> existingRetentionExpiry;
+ if (optionalContext!=null && k.equals(optionalContext.getRetentionHash()) && expiryAccessor.apply(optionalContext.getRetentionSettings())!=null)
+ existingRetentionExpiry = Optional.of(optionalContext);
+ else
+ existingRetentionExpiry = finishedTwins.stream().filter(w -> expiryAccessor.apply(w.getRetentionSettings()) != null).findAny();
+
WorkflowRetentionParser.WorkflowRetentionFilter expiry;
- if (useSoftlyKeptExpiry) {
- expiry = WorkflowRetentionParser.newDefaultSoftFilter().init(finishedTwins.iterator().next());
- } else if (existingRetentionExpiry.isPresent()) {
+
+ 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)) {
- if (optionalContext.getRetentionSettings().expiry != null) {
- if (!optionalContext.getRetentionSettings().expiry.equals(existingRetentionExpiry.get().getRetentionSettings().expiry)) {
- log.warn("Retention specification for " + optionalContext + " '" + optionalContext.getRetentionSettings().expiry + "' is different for same hash. " +
- "Expiry should be constant within a hash but " + existingRetentionExpiry.get() + " has '" + existingRetentionExpiry.get().getRetentionSettings().expiry + "'");
+ // (no such warning for the soft side of it)
+ if (useSoftlyKeptExpiry) {
+ expiry = existingRetentionExpiry.get().getRetentionSettings().getSoftExpiryFn(existingRetentionExpiry.get());
+ } else {
+ if (optionalContext != null && optionalContext.getRetentionHash().equals(k)) {
+ if (optionalContext.getRetentionSettings().expiry != null) {
+ if (!optionalContext.getRetentionSettings().expiry.equals(existingRetentionExpiry.get().getRetentionSettings().expiry)) {
+ log.warn("Retention specification for " + optionalContext + " '" + optionalContext.getRetentionSettings().expiry + "' is different for same hash. " +
+ "Expiry should be constant within a hash but " + existingRetentionExpiry.get() + " has '" + existingRetentionExpiry.get().getRetentionSettings().expiry + "'");
+ }
}
}
+ expiry = existingRetentionExpiry.get().getRetentionSettings().getExpiryFn(existingRetentionExpiry.get());
}
- expiry = existingRetentionExpiry.get().getRetentionSettings().getExpiryFn(existingRetentionExpiry.get());
} else {
- expiry = WorkflowRetentionParser.newDefaultFilter().init(finishedTwins.iterator().next());
+ expiry = WorkflowRetentionParser.newDefaultFilter(useSoftlyKeptExpiry).init(finishedTwins.iterator().next());
}
Collection<WorkflowExecutionContext> retainedFinishedTwins = expiry.apply(finishedTwins);
@@ -169,12 +271,16 @@ public class WorkflowRetentionAndExpiration {
private static boolean isExpirable(WorkflowExecutionContext c) {
if (c.getStatus() == null || !c.getStatus().expirable) return false;
+
+ // should we expire of completed children workflows even if an ancestor workflow is not expirable?
+ // this would prevent silly retention of workflows whose parents are about to end, where the retention check runs on the child just before the parent finishes;
+ // however it would also limit the ability to inspect children workflows e.g. in a foreach block, as only e.g. 3 of the children would be kept ever.
+ // on balance, do NOT expire those; wait for another event to trigger their clean-up.
+ // (if the child workflow is marked disabled, it is not persisted, but takes effect in all other cases.)
if (c.getParent()!=null) {
- // XXX for size reasons, should skip this - fow now, don't expire children workflows unless parents are also expirable
if (!isExpirable(c.getParent())) return false;
-
- // we could weaken this if we have lots of children workflows, but that is more work; left as an enhancement
}
+
return true;
}
@@ -192,6 +298,6 @@ public class WorkflowRetentionAndExpiration {
public static void expireOldWorkflows(Entity entity) {
- new WorkflowStatePersistenceViaSensors(((EntityInternal)entity).getManagementContext()).updateMaps(entity, true, true, null, null);
+ new WorkflowStatePersistenceViaSensors(((EntityInternal)entity).getManagementContext()).updateMaps(entity, null, true, 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 88fe2319f2..0083b167dc 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
@@ -21,6 +21,7 @@ package org.apache.brooklyn.core.workflow.store;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
+import javax.annotation.Nullable;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
@@ -30,7 +31,6 @@ import org.apache.brooklyn.config.ConfigKey;
import org.apache.brooklyn.core.config.ConfigKeys;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
-import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.collections.MutableSet;
import org.slf4j.Logger;
@@ -42,7 +42,10 @@ public class WorkflowStateActiveInMemory {
public static final ConfigKey<WorkflowStateActiveInMemory> IN_MEMORY_WORKFLOWS = ConfigKeys.newConfigKey(WorkflowStateActiveInMemory.class, "internals.brooklyn.workflow.in_memory");
- private static final long GLOBAL_UPDATE_FREQUENCY = 5*60*1000; // every 5m wipe out workflows from old entities
+ long lastInMemEntitiesClear = System.currentTimeMillis();
+
+ // this applies to both sensors and active, but is stored here as this instance is kept on the mgmt context
+ long lastGlobalClear = System.currentTimeMillis();
public static WorkflowStateActiveInMemory get(ManagementContext mgmt) {
WorkflowStateActiveInMemory localActiveWorkflows = mgmt.getScratchpad().get(IN_MEMORY_WORKFLOWS);
@@ -69,10 +72,8 @@ public class WorkflowStateActiveInMemory {
this.mgmt = mgmt;
}
- long lastInMemClear = System.currentTimeMillis();
-
public void expireAbsentEntities() {
- lastInMemClear = System.currentTimeMillis();
+ lastInMemEntitiesClear = System.currentTimeMillis();
Set<String> copy;
synchronized (active) { copy = MutableSet.copyOf(active.keySet()); }
synchronized (completedSoftlyKept) { copy.addAll(completedSoftlyKept.keySet()); }
@@ -89,11 +90,12 @@ public class WorkflowStateActiveInMemory {
if (context.getStatus().expirable) {
withActiveForEntity(context.getEntity().getId(), false, wfm -> wfm.remove(context.getWorkflowId()));
withSoftlyKeptForEntity(context.getEntity().getId(), true, wfm -> { wfm.put(context.getWorkflowId(), context); return null; });
+ recomputeExpiration(context.getEntity(), context);
} else {
// keep active workflows in memory, even if disabled
withActiveForEntity(context.getEntity().getId(), true, wfm -> wfm.put(context.getWorkflowId(), context));
}
- if (lastInMemClear + GLOBAL_UPDATE_FREQUENCY < System.currentTimeMillis()) {
+ if (lastInMemEntitiesClear + WorkflowRetentionAndExpiration.GLOBAL_UPDATE_FREQUENCY < System.currentTimeMillis()) {
// poor man's cleanup, every minute, but good enough
expireAbsentEntities();
}
@@ -160,9 +162,9 @@ public class WorkflowStateActiveInMemory {
return result;
}
- public void recomputeExpiration(Entity entity) {
+ public void recomputeExpiration(Entity entity, @Nullable WorkflowExecutionContext optionalContext) {
withSoftlyKeptForEntity(entity.getId(), false, wfm -> {
- WorkflowRetentionAndExpiration.recomputeExpiration(wfm.asMap(), null, true);
+ WorkflowRetentionAndExpiration.recomputeExpiration(wfm.asMap(), optionalContext, 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 97c37fc1a3..ffd1698c91 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
@@ -18,6 +18,14 @@
*/
package org.apache.brooklyn.core.workflow.store;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
import com.google.common.reflect.TypeToken;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ManagementContext;
@@ -25,7 +33,6 @@ 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.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.sensor.Sensors;
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
@@ -35,15 +42,6 @@ import org.apache.brooklyn.util.guava.Maybe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
public class WorkflowStatePersistenceViaSensors {
private static final Logger log = LoggerFactory.getLogger(WorkflowStatePersistenceViaSensors.class);
@@ -52,8 +50,6 @@ public class WorkflowStatePersistenceViaSensors {
public static final AttributeSensor<Map<String,WorkflowExecutionContext>> INTERNAL_WORKFLOWS = Sensors.newSensor(new TypeToken<Map<String, WorkflowExecutionContext>>() {}, "internals.brooklyn.workflow");
- private static final long GLOBAL_UPDATE_FREQUENCY = 5*60*1000; // every 5m wipe out old workflows
-
public static WorkflowStatePersistenceViaSensors get(ManagementContext mgmt) {
WorkflowStatePersistenceViaSensors sharedInstance = mgmt.getScratchpad().get(SENSOR_WORKFLOW_PERSISTER);
if (sharedInstance==null) {
@@ -76,27 +72,11 @@ public class WorkflowStatePersistenceViaSensors {
enum PersistenceWithQueuedTasks { ALLOW, WARN, FAIL }
- long lastInMemClear = System.currentTimeMillis();
-
- public void checkpoint(WorkflowExecutionContext context) {
- checkpoint(context, PersistenceWithQueuedTasks.WARN);
- }
- public void checkpoint(WorkflowExecutionContext context, PersistenceWithQueuedTasks expectQueuedTasks) {
- doGlobalUpdateIfNeeded();
-
- Entity entity = context.getEntity();
- if (Entities.isUnmanagingOrNoLongerManaged(entity)) {
- log.debug("Skipping persistence of "+context+" as entity is no longer active here");
- return;
- }
-
- // keep active workflows in memory, even if disabled
- WorkflowStateActiveInMemory.get(context.getManagementContext()).checkpoint(context);
-
+ protected void checkpoint(WorkflowExecutionContext context, PersistenceWithQueuedTasks expectQueuedTasks) {
if (Boolean.TRUE.equals(context.getRetentionSettings().disabled)) {
if (getFromTag(BrooklynTaskTags.tagForWorkflow(context), false, false)!=null) {
// need to clear
- updateMap(entity, false, true, v -> v.remove(context.getWorkflowId(), context));
+ updateMap(context.getEntity(), context, false, true, v -> v.remove(context.getWorkflowId(), context));
}
return;
}
@@ -110,28 +90,15 @@ public class WorkflowStatePersistenceViaSensors {
}
}
- expireOldWorkflows(entity, context);
- }
-
- private void doGlobalUpdateIfNeeded() {
- if (lastInMemClear + GLOBAL_UPDATE_FREQUENCY > System.currentTimeMillis()) return;
- lastInMemClear = System.currentTimeMillis();
- AtomicInteger total = new AtomicInteger(0);
- Collection<Entity> entities = mgmt.getEntityManager().getEntities();
- entities.forEach(entity -> {
- int change = expireOldWorkflows(entity, null);
- if (change!=0) log.debug("Global entity workflow persistence update, removed "+(-change)+" workflows from "+entity);
- total.addAndGet(change);
- });
- if (total.get()!=0) log.debug("Global entity workflow persistence update, removed "+(-total.get())+" workflows across all "+entities.size()+" entities");
+ expireOldWorkflowsOnDisk(context.getEntity(), context);
}
- public int expireOldWorkflows(Entity entity, @Nullable WorkflowExecutionContext context) {
+ int expireOldWorkflowsOnDisk(Entity entity, @Nullable WorkflowExecutionContext context) {
// clear interrupt status so we can persist e.g. if we are interrupted or shutdown
boolean interrupted = Thread.interrupted();
boolean doExpiry = WorkflowRetentionAndExpiration.isExpirationCheckNeeded(entity);
try {
- return updateMaps(entity, doExpiry, true, context==null ? null : v -> v.put(context.getWorkflowId(), context), null);
+ return updateMaps(entity, null, doExpiry, false, true, context==null ? null : v -> v.put(context.getWorkflowId(), context), null);
} finally {
if (interrupted) Thread.currentThread().interrupt();
@@ -142,7 +109,7 @@ public class WorkflowStatePersistenceViaSensors {
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);
- updateMaps(w.getEntity(), false, true, map -> {
+ updateMaps(w.getEntity(), w, false, false, true, map -> {
boolean removed = WorkflowRetentionAndExpiration.deleteWorkflowFromMap(map, w, true, true);
if (removed) result.set(true);
}, w);
@@ -153,26 +120,26 @@ public class WorkflowStatePersistenceViaSensors {
}
}
- int updateMaps(Entity entity, boolean doExpiry, boolean persist, Consumer<Map<String,WorkflowExecutionContext>> action, WorkflowExecutionContext contextToRemoveFromSoftMemory) {
- int result = updateMap(entity, doExpiry, persist, action);
+ int updateMaps(Entity entity, @Nullable WorkflowExecutionContext optionalContext, boolean doExpiryForSensor, boolean doExpiryInMemory, boolean persist, Consumer<Map<String,WorkflowExecutionContext>> action, WorkflowExecutionContext contextToRemoveFromSoftMemory) {
+ int result = updateMap(entity, optionalContext, doExpiryForSensor, persist, action);
// and update softly kept
WorkflowStateActiveInMemory activeInMemory = WorkflowStateActiveInMemory.get(mgmt);
if (contextToRemoveFromSoftMemory!=null) {
activeInMemory.deleteWorkflow(contextToRemoveFromSoftMemory);
}
- if (doExpiry) activeInMemory.recomputeExpiration(entity);
+ if (doExpiryInMemory) activeInMemory.recomputeExpiration(entity, optionalContext);
return result;
}
- int updateMap(Entity entity, boolean doExpiry, boolean persist, Consumer<Map<String,WorkflowExecutionContext>> action) {
+ int updateMap(Entity entity, @Nullable WorkflowExecutionContext optionalContext, 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, false);
+ if (doExpiry) v = WorkflowRetentionAndExpiration.recomputeExpiration(v, optionalContext, false);
delta.getAndAdd(v.size());
return Maybe.of(v);
});
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 a5dbe86b0f..db5a2bdf3f 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
@@ -18,6 +18,18 @@
*/
package org.apache.brooklyn.core.workflow.utils;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
import org.apache.brooklyn.core.workflow.WorkflowExpressionResolution;
import org.apache.brooklyn.core.workflow.store.WorkflowRetentionAndExpiration;
@@ -27,13 +39,7 @@ import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.guava.Maybe;
import org.apache.brooklyn.util.text.Strings;
import org.apache.brooklyn.util.time.Duration;
-
-import javax.annotation.Nullable;
-import java.time.Instant;
-import java.time.temporal.ChronoUnit;
-import java.util.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
+import org.apache.commons.lang3.tuple.Pair;
public class WorkflowRetentionParser {
@@ -82,17 +88,48 @@ also allows `hard` at start or end, or `soft [limit]` at end
continue;
}
- // TODO soft/hard keyword; take whichever occurs last
- if (retentionExpression.contains(" hash ")) {
- if (result.hash != null)
- throw new IllegalArgumentException("Cannot set multiple 'hash' in retention expression");
- int i = retentionExpression.indexOf(" hash ");
- result.hash = Strings.removeFromStart(retentionExpression.substring(i).trim(), "hash").trim();
- retentionExpression = retentionExpression.substring(0, i).trim();
- continue;
+ List<Pair<String,Integer>> specialTerms = MutableList.of();
+ for (String term: MutableList.of("hash", "soft", "hard"))
+ specialTerms.add(Pair.of(term, retentionExpression.indexOf(" "+term+" ")));
+ Collections.sort(specialTerms, (x,y) -> -Integer.compare(x.getRight(), y.getRight()));
+ Pair<String, Integer> last = specialTerms.iterator().next();
+ if (last.getRight()>=0) {
+ if ("hash".equals(last.getLeft())) {
+ if (result.hash != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hash' in retention expression");
+ result.hash = Strings.removeFromStart(retentionExpression.substring(last.getRight()).trim(), last.getLeft()).trim();
+ retentionExpression = retentionExpression.substring(0, last.getRight()).trim();
+ continue;
+ }
+ if ("hard".equals(last.getLeft())) {
+ if (result.softExpiry != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hard' or 'soft' in retention expression");
+ result.softExpiry = "0";
+ String hardTrailing = Strings.removeFromStart(retentionExpression.substring(last.getRight()).trim(), last.getLeft()).trim();
+ if (Strings.isNonBlank(hardTrailing)) {
+ if (last.getRight() == 0) retentionExpression = hardTrailing;
+ else throw new IllegalArgumentException("Cannot have retention definition both before and after 'hard' keyword");
+ } else {
+ retentionExpression = retentionExpression.substring(0, last.getRight()).trim();
+ }
+ continue;
+ }
+ if ("soft".equals(last.getLeft())) {
+ if (result.softExpiry != null)
+ throw new IllegalArgumentException("Cannot set multiple 'hard' or 'soft' in retention expression");
+ String softTrailing = Strings.removeFromStart(retentionExpression.substring(last.getRight()).trim(), last.getLeft()).trim();
+ if (Strings.isNonBlank(softTrailing)) {
+ result.softExpiry = softTrailing;
+ new WorkflowRetentionParser(result.softExpiry).soft().parse();
+ retentionExpression = retentionExpression.substring(0, last.getRight()).trim();
+ } else {
+ throw new IllegalArgumentException("Specification for 'soft' retetntion must provide retention expression after the keyword");
+ }
+ continue;
+ }
}
break;
- } while (false);
+ } while (true);
if (retentionExpression.equals("disabled")) {
result.disabled = true;
@@ -201,6 +238,8 @@ also allows `hard` at start or end, or `soft [limit]` at end
static abstract class KeepDelegate implements WorkflowRetentionFilter {
WorkflowRetentionFilter delegate;
+ final boolean soft;
+ KeepDelegate(boolean soft) { this.soft = soft; }
@Override
public Collection<WorkflowExecutionContext> apply(Collection<WorkflowExecutionContext> workflowExecutionContexts) {
if (delegate==null) throw new IllegalStateException("Not initialized");
@@ -214,30 +253,31 @@ also allows `hard` at start or end, or `soft [limit]` at end
protected abstract WorkflowRetentionFilter findDelegate(WorkflowExecutionContext workflow);
}
static class KeepSystem extends KeepDelegate {
+ KeepSystem(boolean soft) { super(soft); }
@Override
public WorkflowRetentionFilter findDelegate(WorkflowExecutionContext workflow) {
if (workflow==null) throw new IllegalStateException("Retention 'system' cannot be used here");
- return new WorkflowRetentionParser(workflow.getManagementContext().getConfig().getConfig(WorkflowRetentionAndExpiration.WORKFLOW_RETENTION_DEFAULT)).parse().init(null);
+ return new WorkflowRetentionParser(workflow.getManagementContext().getConfig().getConfig(
+ soft ? WorkflowRetentionAndExpiration.WORKFLOW_RETENTION_DEFAULT_SOFT : WorkflowRetentionAndExpiration.WORKFLOW_RETENTION_DEFAULT))
+ .soft(soft).parse().init(null);
}
@Override
public String toString() {
return "system";
}
}
- public static WorkflowRetentionFilter newDefaultFilter() {
- return new KeepParent();
- }
- public static WorkflowRetentionFilter newDefaultSoftFilter() {
- return new KeepSystem();
+ public static WorkflowRetentionFilter newDefaultFilter(boolean soft) {
+ return new KeepParent(soft);
}
static class KeepParent extends KeepDelegate {
+ KeepParent(boolean soft) { super(soft); }
@Override
public WorkflowRetentionFilter findDelegate(WorkflowExecutionContext workflow) {
if (workflow == null) throw new IllegalStateException("Retention 'parent' cannot be used here");
else if (workflow.getParent()!=null) {
- return workflow.getParent().getRetentionSettings().getExpiryFn(workflow.getParent());
+ return soft ? workflow.getParent().getRetentionSettings().getSoftExpiryFn(workflow.getParent()) : workflow.getParent().getRetentionSettings().getExpiryFn(workflow.getParent());
} else {
- return new KeepSystem().init(workflow);
+ return new KeepSystem(soft).init(workflow);
}
}
@Override
@@ -246,12 +286,13 @@ also allows `hard` at start or end, or `soft [limit]` at end
}
}
static class KeepContext extends KeepDelegate {
+ KeepContext(boolean soft) { super(soft); }
@Override
public WorkflowRetentionFilter findDelegate(WorkflowExecutionContext workflow) {
if (workflow == null) throw new IllegalStateException("Retention 'context' cannot be used here");
// expands to string to something that doesn't reference context so that this does not infinitely recurse
- return workflow.getRetentionSettings().getExpiryFn(workflow);
+ return soft ? workflow.getRetentionSettings().getSoftExpiryFn(workflow) : workflow.getRetentionSettings().getExpiryFn(workflow);
}
@Override
public String toString() {
@@ -261,17 +302,21 @@ also allows `hard` at start or end, or `soft [limit]` at end
String fullExpression;
String rest;
+ boolean soft = false;
public WorkflowRetentionParser(String fullExpression) {
this.fullExpression = fullExpression;
}
+ public WorkflowRetentionParser soft() { return soft(true); }
+ public WorkflowRetentionParser soft(boolean soft) { this.soft = soft; return this; }
+
public WorkflowRetentionFilter parse() {
- if (Strings.isBlank(fullExpression)) return newDefaultFilter();
+ if (Strings.isBlank(fullExpression)) return newDefaultFilter(soft);
rest = Strings.trimStart(fullExpression.toLowerCase());
WorkflowRetentionFilter result = parseTerm();
- if (!Strings.isBlank(rest)) return newDefaultFilter();
+ if (!Strings.isBlank(rest)) return newDefaultFilter(soft);
return result;
}
@@ -337,9 +382,9 @@ also allows `hard` at start or end, or `soft [limit]` at end
if (term.isPresent()) return term.get();
if (eatNA("all") || eatNA("forever")) return new KeepAll();
- if (eatNA("system")) return new KeepSystem();
- if (eatNA("parent")) return new KeepParent();
- if (eatNA("context")) return new KeepContext();
+ if (eatNA("system")) return new KeepSystem(soft);
+ if (eatNA("parent")) return new KeepParent(soft);
+ if (eatNA("context")) return new KeepContext(soft);
int i = maxPositive(rest.indexOf(","), rest.indexOf(")"));
if (i==-1) i = rest.length();
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 f90a0a27d9..4f5cb3684c 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
@@ -20,6 +20,7 @@ package org.apache.brooklyn.core.workflow;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.google.common.base.Predicates;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import org.apache.brooklyn.api.entity.EntityLocal;
@@ -39,6 +40,7 @@ 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.WorkflowRetentionAndExpiration;
import org.apache.brooklyn.core.workflow.store.WorkflowStatePersistenceViaSensors;
import org.apache.brooklyn.entity.stock.BasicApplication;
import org.apache.brooklyn.test.Asserts;
@@ -927,6 +929,18 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
Asserts.assertSize(wp.get().getWorkflows(app, false).keySet(), 1);
Asserts.assertSize(wp.get().getWorkflows(app, true).keySet(), 3);
+ // setting soft 4 should give one more, and allow hash keyword at end
+ w1 = doTestRetentionDisabled(2, "min(1,context) soft 4 hash my-fixed-hash", false, false, false);
+ Asserts.assertEquals(lastWorkflowContext.getRetentionSettings().softExpiryResolved, "4");
+ Asserts.assertSize(wp.get().getWorkflows(app, false).keySet(), 1);
+ Asserts.assertSize(wp.get().getWorkflows(app, true).keySet(), 4);
+
+ // and soft with min(2,...) allowing hash keyword before should also work
+ w1 = doTestRetentionDisabled(2, "min(1,context) hash my-fixed-hash soft min(2,system)", false, false, false);
+ Asserts.assertEquals(lastWorkflowContext.getRetentionSettings().softExpiryResolved, "min(2,system)");
+ Asserts.assertSize(wp.get().getWorkflows(app, false).keySet(), 1);
+ Asserts.eventually(() -> wp.get().getWorkflows(app, true).keySet().size(), Predicates.equalTo(2), Duration.seconds(2));
+
// 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, false).keySet(), MutableSet.of(w1.getWorkflowId(), w2.getWorkflowId()));
@@ -978,7 +992,7 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
// wait 5s and run something, it should cause everything else to expire
Time.sleep(Duration.FIVE_SECONDS);
- wp.get().expireOldWorkflows(app, null);
+ WorkflowRetentionAndExpiration.expireOldWorkflows(app);
// should now be empty
Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of());
@@ -991,17 +1005,17 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
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);
+ WorkflowRetentionAndExpiration.expireOldWorkflows(app);
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);
+ WorkflowRetentionAndExpiration.expireOldWorkflows(app);
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);
+ WorkflowRetentionAndExpiration.expireOldWorkflows(app);
Asserts.assertEquals(wp.get().getWorkflows(app, false).keySet(), MutableSet.of(w3.getWorkflowId()));
// run two more, that's all we should have
diff --git a/software/base/src/test/java/org/apache/brooklyn/entity/software/base/SoftwareProcessRebindNotRunningEntityTest.java b/software/base/src/test/java/org/apache/brooklyn/entity/software/base/SoftwareProcessRebindNotRunningEntityTest.java
index 7d1849af97..4defc5b109 100644
--- a/software/base/src/test/java/org/apache/brooklyn/entity/software/base/SoftwareProcessRebindNotRunningEntityTest.java
+++ b/software/base/src/test/java/org/apache/brooklyn/entity/software/base/SoftwareProcessRebindNotRunningEntityTest.java
@@ -129,7 +129,13 @@ public class SoftwareProcessRebindNotRunningEntityTest extends RebindTestFixture
}
if (latch instanceof TerminableCountDownLatch) ((TerminableCountDownLatch)latch).terminate();
}
- super.tearDown(Duration.millis(10)); // stops here can be blocked, don't wait on them
+ try {
+ super.tearDown(Duration.millis(50)); // stops here can be blocked, don't wait on them
+ } catch (Exception e) {
+ // we fail on this in case it is a real problem, but not believed to be, only seen occasionally, and not since timeout was increased 2024-04-01
+ LOG.warn("Teardown of test encountered exception; not unknown if multiple processes attempt to destroy, as destruction is deliberately unsynchronized to minimize race errors", e);
+ throw Exceptions.propagateAnnotated("Concurrent teardown issue", e);
+ }
if (executor != null) executor.shutdownNow();
} finally {
latches.clear();
(brooklyn-server) 03/06: expire appropriate softly-kept in memory completed workflows, using system default
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit fb7c01507a1bcb4bf01b48f33923bf161d168642
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Sat Mar 30 01:22:34 2024 +0000
expire appropriate softly-kept in memory completed workflows, using system default
---
.../brooklyn/camp/brooklyn/WorkflowYamlTest.java | 2 +-
.../store/WorkflowRetentionAndExpiration.java | 14 ++++----
.../store/WorkflowStateActiveInMemory.java | 7 ++++
.../store/WorkflowStatePersistenceViaSensors.java | 28 ++++++++++++----
.../workflow/utils/WorkflowRetentionParser.java | 3 ++
.../workflow/WorkflowPersistReplayErrorsTest.java | 37 ++++++++++++++--------
6 files changed, 65 insertions(+), 26 deletions(-)
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 b7ba499072..a4dc8d3bc1 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 @@ public class WorkflowYamlTest extends AbstractYamlTest {
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 c6c8a16d73..4c2c667d59 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 @@ public class WorkflowRetentionAndExpiration {
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 @@ public class WorkflowRetentionAndExpiration {
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 @@ public class WorkflowRetentionAndExpiration {
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 @@ public class WorkflowRetentionAndExpiration {
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 class WorkflowRetentionAndExpiration {
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 59f5780a87..88fe2319f2 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 @@ public class WorkflowStateActiveInMemory {
}
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 82ef7b0c30..97c37fc1a3 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 @@ public class WorkflowStatePersistenceViaSensors {
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 @@ public class WorkflowStatePersistenceViaSensors {
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 @@ public class WorkflowStatePersistenceViaSensors {
}
}
+ 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 class WorkflowStatePersistenceViaSensors {
}
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 d2488457bc..a5dbe86b0f 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 @@ also allows `hard` at start or end, or `soft [limit]` at end
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 adc6fd2702..f90a0a27d9 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 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
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 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
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 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
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 @@ public class WorkflowPersistReplayErrorsTest extends RebindTestFixture<BasicAppl
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;
(brooklyn-server) 01/06: record the workflow resolution context better
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit 931caf0bf03e306d7cd15e296a7ffdcb6ef04b5a
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Fri Mar 29 16:27:06 2024 +0000
record the workflow resolution context better
allows better tracing, and the ability to use that context when coercing
---
.../workflow/WorkflowExpressionResolution.java | 247 ++++++++++++---------
.../brooklyn/util/core/text/TemplateProcessor.java | 6 +-
.../util/collections/ThreadLocalStack.java | 23 +-
3 files changed, 163 insertions(+), 113 deletions(-)
diff --git a/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExpressionResolution.java b/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExpressionResolution.java
index c8e32abad5..658aeb1b4d 100644
--- a/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExpressionResolution.java
+++ b/core/src/main/java/org/apache/brooklyn/core/workflow/WorkflowExpressionResolution.java
@@ -20,7 +20,6 @@ package org.apache.brooklyn.core.workflow;
import java.time.Instant;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -29,6 +28,9 @@ import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import javax.annotation.Nullable;
import com.google.common.annotations.Beta;
import com.google.common.reflect.TypeToken;
@@ -188,7 +190,7 @@ public class WorkflowExpressionResolution {
}
} catch (Throwable t) {
Exceptions.propagateIfFatal(t);
- if (stage==WorkflowExpressionStage.STEP_INPUT && WorkflowVariableResolutionStackEntry.isStackForSettingVariable(RESOLVE_STACK.getAll(true), key) && Exceptions.getFirstThrowableOfType(t, WorkflowVariableRecursiveReference.class)!=null) {
+ if (stage==WorkflowExpressionStage.STEP_INPUT && isSettingVariable(key) && Exceptions.getFirstThrowableOfType(t, WorkflowVariableRecursiveReference.class)!=null) {
// input evaluation can look at local input, and will gracefully handle some recursive references.
// this is needed so we can handle things like env:=${env} in input, and also {message:="Hi ${name}", name:="Bob"}.
@@ -401,94 +403,119 @@ public class WorkflowExpressionResolution {
/** does not use templates */
public <T> T resolveCoercingOnly(Object expression, TypeToken<T> type) {
if (expression==null) return null;
- boolean triedCoercion = false;
- List<Exception> exceptions = MutableList.of();
- if (expression instanceof String) {
- try {
- // prefer simple coercion if it's a string coming in
- return TypeCoercions.coerce(expression, type);
- } catch (Exception e) {
- Exceptions.propagateIfFatal(e);
- exceptions.add(e);
- triedCoercion = true;
+ return inResolveStackEntry("resolve-coercing", expression, () -> {
+ boolean triedCoercion = false;
+ List<Exception> exceptions = MutableList.of();
+ if (expression instanceof String) {
+ try {
+ // prefer simple coercion if it's a string coming in
+ return TypeCoercions.coerce(expression, type);
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ exceptions.add(e);
+ triedCoercion = true;
+ }
}
- }
- if (Jsonya.isJsonPrimitiveDeep(expression) && !(expression instanceof Set)) {
- try {
- // next try yaml coercion for anything complex, as values are normally set from yaml and will be raw at this stage (but not if they are from a DSL)
- return BeanWithTypeUtils.convert(context.getManagementContext(), expression, type, true,
- RegisteredTypes.getClassLoadingContext(context.getEntity()), true /* needed for wrapped resolved holders */);
- } catch (Exception e) {
- Exceptions.propagateIfFatal(e);
- exceptions.add(e);
+ if (Jsonya.isJsonPrimitiveDeep(expression) && !(expression instanceof Set)) {
+ try {
+ // next try yaml coercion for anything complex, as values are normally set from yaml and will be raw at this stage (but not if they are from a DSL)
+ return BeanWithTypeUtils.convert(context.getManagementContext(), expression, type, true,
+ RegisteredTypes.getClassLoadingContext(context.getEntity()), true /* needed for wrapped resolved holders */);
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ exceptions.add(e);
+ }
}
- }
- if (!triedCoercion) {
- try {
- // fallback to simple coercion
- return TypeCoercions.coerce(expression, type);
- } catch (Exception e) {
- Exceptions.propagateIfFatal(e);
- exceptions.add(e);
- triedCoercion = true;
+ if (!triedCoercion) {
+ try {
+ // fallback to simple coercion
+ return TypeCoercions.coerce(expression, type);
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ exceptions.add(e);
+ triedCoercion = true;
+ }
}
- }
- throw Exceptions.propagate(exceptions.iterator().next());
+ throw Exceptions.propagate(exceptions.iterator().next());
+ });
}
- static class WorkflowVariableResolutionStackEntry {
+ public static class WorkflowResolutionStackEntry {
+ // resolver is null if caller has indicated evaluation before resolution
+ @Nullable WorkflowExpressionResolution resolver;
WorkflowExecutionContext context;
WorkflowExpressionStage stage;
- Object object;
+ String callPointUid;
+ Object expression;
String settingVariable;
- public static WorkflowVariableResolutionStackEntry of(WorkflowExecutionContext context, WorkflowExpressionStage stage, Object expression) {
- WorkflowVariableResolutionStackEntry result = new WorkflowVariableResolutionStackEntry();
+ public static WorkflowResolutionStackEntry of(WorkflowExpressionResolution resolver, String callPointUid, Object expression) {
+ WorkflowResolutionStackEntry result = of(resolver == null ? null : resolver.context, resolver.stage, callPointUid, expression);
+ result.resolver = resolver;
+ return result;
+ }
+ public static WorkflowResolutionStackEntry of(WorkflowExecutionContext context, WorkflowExpressionStage stage, String callPointUid, Object expression) {
+ WorkflowResolutionStackEntry result = new WorkflowResolutionStackEntry();
result.context = context;
+ result.callPointUid = callPointUid;
result.stage = stage;
- result.object = expression;
+ result.expression = expression;
return result;
}
- public static WorkflowVariableResolutionStackEntry setting(WorkflowExecutionContext context, WorkflowExpressionStage stage, String settingVariable) {
- WorkflowVariableResolutionStackEntry result = new WorkflowVariableResolutionStackEntry();
- result.context = context;
- result.stage = stage;
+ public static WorkflowResolutionStackEntry settingVariable(WorkflowExecutionContext context, WorkflowExpressionStage stage, String settingVariable) {
+ WorkflowResolutionStackEntry result = of(context, stage, "setting-variable", null);
result.settingVariable = settingVariable;
return result;
}
- public static boolean isStackForSettingVariable(Collection<WorkflowVariableResolutionStackEntry> stack, String key) {
+ public static boolean isStackForSettingVariable(Stream<WorkflowResolutionStackEntry> stack, String key) {
if (stack==null) return true;
- MutableList<WorkflowVariableResolutionStackEntry> s2 = MutableList.copyOf(stack);
- Collections.reverse(s2);
- Optional<WorkflowVariableResolutionStackEntry> s = s2.stream().filter(si -> si.settingVariable != null).findFirst();
+ Optional<WorkflowResolutionStackEntry> s = stack.filter(si -> si.settingVariable != null).findFirst();
if (!s.isPresent()) return false;
return s.get().settingVariable.equals(key);
}
+ public String getWorkflowId() {
+ WorkflowExecutionContext ctx = getWorkflowExecutionContext();
+ return ctx == null ? null : ctx.getWorkflowId();
+ }
+ public WorkflowExpressionResolution getWorkflowExpressionResolution() {
+ return resolver;
+ }
+ public WorkflowExecutionContext getWorkflowExecutionContext() {
+ return context;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- WorkflowVariableResolutionStackEntry that = (WorkflowVariableResolutionStackEntry) o;
+ WorkflowResolutionStackEntry that = (WorkflowResolutionStackEntry) o;
- if (context != null && that.context != null ? !Objects.equals(context.getWorkflowId(), that.context.getWorkflowId()) : !Objects.equals(context, that.context)) return false;
+ String wid = getWorkflowId();
+ String tid = that.getWorkflowId();
+ // might have different contexts with same ID; but if ID not set for some reason then use context
+ boolean checkIdNotContext = wid!=null && tid!=null;
+ if (checkIdNotContext && tid!=null && !Objects.equals(wid, tid)) return false;
+ if (!checkIdNotContext && !Objects.equals(getWorkflowExecutionContext(), that.getWorkflowExecutionContext())) return false;
if (stage != that.stage) return false;
- if (object != null ? !object.equals(that.object) : that.object != null) return false;
+ if (!Objects.equals(callPointUid, that.callPointUid)) return false;
+ if (expression != null ? !expression.equals(that.expression) : that.expression != null) return false;
if (settingVariable != null ? !settingVariable.equals(that.settingVariable) : that.settingVariable != null) return false;
return true;
}
@Override
public int hashCode() {
- int result = context != null && context.getWorkflowId()!=null ? context.getWorkflowId().hashCode() : 0;
+ int result = getWorkflowId() != null ? getWorkflowId().hashCode() : 0;
result = 31 * result + (stage != null ? stage.hashCode() : 0);
- result = 31 * result + (object != null ? object.hashCode() : 0);
+ result = 31 * result + (callPointUid != null ? callPointUid.hashCode() : 0);
+ result = 31 * result + (expression != null ? expression.hashCode() : 0);
result = 31 * result + (settingVariable != null ? settingVariable.hashCode() : 0);
return result;
}
@@ -497,27 +524,39 @@ public class WorkflowExpressionResolution {
/** method which can be used to indicate that a reference to the variable, if it is recursive, is recoverable, because we are in the process of setting that variable.
* see discussion on usages of WorkflowVariableResolutionStackEntry.isStackForSettingVariable */
public static <T> T allowingRecursionWhenSetting(WorkflowExecutionContext context, WorkflowExpressionStage stage, String variable, Supplier<T> callable) {
- WorkflowVariableResolutionStackEntry entry = null;
- try {
- entry = WorkflowVariableResolutionStackEntry.setting(context, stage, variable);
- if (!RESOLVE_STACK.push(entry)) {
- entry = null;
- throw new WorkflowVariableRecursiveReference("Recursive or missing reference setting "+variable+": "+RESOLVE_STACK.getAll(false).stream().map(p -> p.object!=null ? p.object.toString() : p.settingVariable).collect(Collectors.joining("->")));
- }
+ return inResolveStackEntry(WorkflowResolutionStackEntry.settingVariable(context, stage, variable), () -> {
+ throw new WorkflowVariableRecursiveReference("Recursive or missing reference setting "+variable+": "+RESOLVE_STACK.stream().map(p -> p.expression !=null ? p.expression.toString() : p.settingVariable).filter(x -> x!=null).collect(Collectors.joining("->")));
+ },
+ callable);
+ }
- return callable.get();
+ static ThreadLocalStack<WorkflowResolutionStackEntry> RESOLVE_STACK = new ThreadLocalStack<>(false);
+ <T> T inResolveStackEntry(String callPointUid, Object expression, Supplier<T> code) {
+ return inResolveStackEntry(WorkflowResolutionStackEntry.of(this, callPointUid, expression), null, code);
+ }
+ static <T> T inResolveStackEntry(WorkflowResolutionStackEntry entry, Runnable errorIfDuplicate, Supplier<T> code) {
+ boolean added = RESOLVE_STACK.push(entry);
+ if (!added && errorIfDuplicate!=null) errorIfDuplicate.run();
+ try {
+ return code.get();
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
} finally {
- if (entry!=null) {
- RESOLVE_STACK.pop(entry);
- }
+ if (added) RESOLVE_STACK.pop(entry);
}
}
- static ThreadLocalStack<WorkflowVariableResolutionStackEntry> RESOLVE_STACK = new ThreadLocalStack<>(false);
-
WorkflowExpressionStage previousStage() {
- return RESOLVE_STACK.peekPenultimate().map(s -> s.stage).orNull();
+ return RESOLVE_STACK.stream().skip(1).map(s -> s.stage).filter(s -> s!=null).findFirst().orElse(null);
+ }
+
+ public static boolean isSettingVariable(String key) {
+ return WorkflowResolutionStackEntry.isStackForSettingVariable(RESOLVE_STACK.stream(), key);
+ }
+
+ public static WorkflowExpressionResolution getCurrentWorkflowExpressionResolution() {
+ return RESOLVE_STACK.stream().map(WorkflowResolutionStackEntry::getWorkflowExpressionResolution).filter(x -> x!=null).findFirst().orElse(null);
}
public static class WorkflowVariableRecursiveReference extends IllegalArgumentException {
@@ -547,50 +586,48 @@ public class WorkflowExpressionResolution {
}
public Object processTemplateExpression(Object expression, AllowBrooklynDslMode allowBrooklynDsl) {
- WorkflowVariableResolutionStackEntry entry = null;
- try {
- entry = WorkflowVariableResolutionStackEntry.of(context, stage, expression);
- if (!RESOLVE_STACK.push(entry)) {
- entry = null;
- throw new WorkflowVariableRecursiveReference("Recursive reference: " + RESOLVE_STACK.getAll(false).stream().map(p -> "" + p.object).collect(Collectors.joining("->")));
- }
- if (RESOLVE_STACK.size() > 100) {
- throw new WorkflowVariableRecursiveReference("Reference exceeded max depth 100: " + RESOLVE_STACK.getAll(false).stream().map(p -> "" + p.object).collect(Collectors.joining("->")));
- }
+ return inResolveStackEntry(WorkflowResolutionStackEntry.of(this, "process-template-expression", expression), () -> {
+ throw new WorkflowVariableRecursiveReference("Recursive reference: " + RESOLVE_STACK.getAll(false).stream().map(p -> "" + p.expression).collect(Collectors.joining("->")));
+ }, () -> {
+ try {
+ if (RESOLVE_STACK.size() > 100) {
+ throw new WorkflowVariableRecursiveReference("Reference exceeded max depth 100: " + RESOLVE_STACK.getAll(false).stream().map(p -> "" + p.expression).collect(Collectors.joining("->")));
+ }
- Object result;
- if (expression instanceof String) result = processTemplateExpressionString((String) expression, allowBrooklynDsl);
- else if (expression instanceof Map) result = processTemplateExpressionMap((Map) expression, allowBrooklynDsl);
- else if (expression instanceof Collection)
- result = processTemplateExpressionCollection((Collection) expression, allowBrooklynDsl);
- else if (expression == null || Boxing.isPrimitiveOrBoxedObject(expression)) result = expression;
- else {
- // otherwise resolve DSL
- result = allowBrooklynDsl.isAllowedHere() ? resolveDsl(expression) : expression;
- if (wrappingMode.wrapResolvedValues && !Objects.equals(result, expression) && !(result instanceof DeferredSupplier)) {
- result = WrappedResolvedExpression.ifNonDeferred(expression, result);
+ Object result;
+ if (expression instanceof String)
+ result = processTemplateExpressionString((String) expression, allowBrooklynDsl);
+ else if (expression instanceof Map)
+ result = processTemplateExpressionMap((Map) expression, allowBrooklynDsl);
+ else if (expression instanceof Collection)
+ result = processTemplateExpressionCollection((Collection) expression, allowBrooklynDsl);
+ else if (expression == null || Boxing.isPrimitiveOrBoxedObject(expression)) result = expression;
+ else {
+ // otherwise resolve DSL
+ result = allowBrooklynDsl.isAllowedHere() ? resolveDsl(expression) : expression;
+ if (wrappingMode.wrapResolvedValues && !Objects.equals(result, expression) && !(result instanceof DeferredSupplier)) {
+ result = WrappedResolvedExpression.ifNonDeferred(expression, result);
+ }
}
- }
- return result;
+ return result;
- } catch (Exception e) {
- Exception e2 = e;
- if (wrappingMode.deferAndRetryErroneousExpressions) {
- return WrappedUnresolvedExpression.ofExpression(expression, this, allowBrooklynDsl);
- }
- if (!allowWaiting && Exceptions.isCausedByInterruptInAnyThread(e)) {
- e2 = new IllegalArgumentException("Expression value '"+expression+"' unavailable and not permitted to wait: "+ Exceptions.collapseText(e), e);
- }
- if (wrappingMode.deferThrowingError) {
- // in wrapped value mode, errors don't throw until accessed, and when used in conditions they can be tested as absent
- return WrappedResolvedExpression.ofError(expression, new ResolutionFailureTreatedAsAbsent.ResolutionFailureTreatedAsAbsentDefaultException(e2));
- } else {
- throw Exceptions.propagate(e2);
+ } catch (Exception e) {
+ Exception e2 = e;
+ if (wrappingMode.deferAndRetryErroneousExpressions) {
+ return WrappedUnresolvedExpression.ofExpression(expression, this, allowBrooklynDsl);
+ }
+ if (!allowWaiting && Exceptions.isCausedByInterruptInAnyThread(e)) {
+ e2 = new IllegalArgumentException("Expression value '" + expression + "' unavailable and not permitted to wait: " + Exceptions.collapseText(e), e);
+ }
+ if (wrappingMode.deferThrowingError) {
+ // in wrapped value mode, errors don't throw until accessed, and when used in conditions they can be tested as absent
+ return WrappedResolvedExpression.ofError(expression, new ResolutionFailureTreatedAsAbsent.ResolutionFailureTreatedAsAbsentDefaultException(e2));
+ } else {
+ throw Exceptions.propagate(e2);
+ }
}
- } finally {
- if (entry != null) RESOLVE_STACK.pop(entry);
- }
+ });
}
private Object resolveDsl(Object expression) {
@@ -631,7 +668,7 @@ public class WorkflowExpressionResolution {
return errorMode;
}
- public Object processTemplateExpressionString(String expression, AllowBrooklynDslMode allowBrooklynDsl) {
+ protected Object processTemplateExpressionString(String expression, AllowBrooklynDslMode allowBrooklynDsl) {
Object result;
boolean ourWait = false;
try {
@@ -698,7 +735,7 @@ public class WorkflowExpressionResolution {
interruptSetIfNeededToPreventWaiting.remove();
}
- public Object processTemplateExpressionMap(Map<?,?> object, AllowBrooklynDslMode allowBrooklynDsl) {
+ protected Object processTemplateExpressionMap(Map<?,?> object, AllowBrooklynDslMode allowBrooklynDsl) {
if (allowBrooklynDsl.isAllowedHere() && object.size()==1) {
Object key = object.keySet().iterator().next();
if (key instanceof String && ((String)key).startsWith("$brooklyn:")) {
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/text/TemplateProcessor.java b/core/src/main/java/org/apache/brooklyn/util/core/text/TemplateProcessor.java
index 8a51d7c1b4..af00cbce16 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/text/TemplateProcessor.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/text/TemplateProcessor.java
@@ -86,7 +86,7 @@ public class TemplateProcessor {
public static TemplateModel wrapAsTemplateModel(Object o) throws TemplateModelException { return BROOKLYN_WRAPPER.wrap(o); }
public static Maybe<Object> unwrapTemplateModelMaybe(TemplateModel templateModel) { return BROOKLYN_WRAPPER.unwrapMaybe(templateModel); }
- static ThreadLocalStack<Map<TemplateModel,Object>> TEMPLATE_MODEL_UNWRAP_CACHE = new ThreadLocalStack<>(true);
+ static ThreadLocalStack<Map<TemplateModel,Object>> TEMPLATE_MODEL_UNWRAP_CACHE = new ThreadLocalStack<>();
/** A cache is used to be able to retrieve the object from which a TemplateModel was created, if needed,
* because Freemarker doesn't support that except on selected UnwrappableTemplateModel subclasses.
* Use wrap and unwrap methods above to access.
@@ -94,8 +94,8 @@ public class TemplateProcessor {
public static void openLocalTemplateModelCache() { TEMPLATE_MODEL_UNWRAP_CACHE.push(MutableMap.of()); }
public static void closeLocalTemplateModelCache() { TEMPLATE_MODEL_UNWRAP_CACHE.pop(); }
- static ThreadLocalStack<String> TEMPLATE_FILE_WANTING_LEGACY_SYNTAX = new ThreadLocalStack<>(true);
- static ThreadLocalStack<Boolean> IS_FOR_WORKFLOW = new ThreadLocalStack<>(true);
+ static ThreadLocalStack<String> TEMPLATE_FILE_WANTING_LEGACY_SYNTAX = new ThreadLocalStack<>();
+ static ThreadLocalStack<Boolean> IS_FOR_WORKFLOW = new ThreadLocalStack<>();
public interface UnwrappableTemplateModel {
Maybe<Object> unwrap();
diff --git a/utils/common/src/main/java/org/apache/brooklyn/util/collections/ThreadLocalStack.java b/utils/common/src/main/java/org/apache/brooklyn/util/collections/ThreadLocalStack.java
index d225e1770f..f9a3f90479 100644
--- a/utils/common/src/main/java/org/apache/brooklyn/util/collections/ThreadLocalStack.java
+++ b/utils/common/src/main/java/org/apache/brooklyn/util/collections/ThreadLocalStack.java
@@ -22,23 +22,27 @@ import com.google.common.collect.Iterables;
import org.apache.brooklyn.util.guava.Maybe;
import java.util.Collection;
+import java.util.Collections;
import java.util.Iterator;
import java.util.Objects;
+import java.util.stream.Stream;
public class ThreadLocalStack<T> implements Iterable<T> {
- private final boolean allowDuplicates;
+ private final boolean acceptDuplicates;
- public ThreadLocalStack(boolean allowsDuplicates) {
- this.allowDuplicates = allowsDuplicates;
+ /** if duplicates not accepted, the call to push will return false */
+ public ThreadLocalStack(boolean acceptDuplicates) {
+ this.acceptDuplicates = acceptDuplicates;
}
+ public ThreadLocalStack() { this.acceptDuplicates = true; }
final ThreadLocal<Collection<T>> set = new ThreadLocal<>();
public Collection<T> getAll(boolean forceInitialized) {
Collection<T> result = set.get();
if (forceInitialized && result==null) {
- result = allowDuplicates ? MutableList.of() : MutableSet.of();
+ result = acceptDuplicates ? MutableList.of() : MutableSet.of();
set.set(result);
}
return result;
@@ -52,13 +56,22 @@ public class ThreadLocalStack<T> implements Iterable<T> {
return last;
}
+ /** returns true unless duplicates are not accepted, in which case it returns false iff the object supplied is equal to one already present */
public boolean push(T object) {
return getAll(true).add(object);
}
+ /** top of stack first */
@Override
public Iterator<T> iterator() {
- return null;
+ return stream().iterator();
+ }
+
+ /** top of stack first */
+ public Stream<T> stream() {
+ MutableList<T> l = MutableList.copyOf(getAll(false));
+ Collections.reverse(l);
+ return l.stream();
}
public Maybe<T> peek() {
(brooklyn-server) 05/06: pause most tasks submitted during entity startup, until all entities are managed
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heneveld pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brooklyn-server.git
commit 9ef5dcc7a40add81414786fa2be84e19084f71f3
Author: Alex Heneveld <al...@cloudsoft.io>
AuthorDate: Mon Apr 1 15:29:15 2024 +0100
pause most tasks submitted during entity startup, until all entities are managed
can affect new tasks and especially rebind if an adjunct is started and tries to access another entity
which hasn't yet started up; particularly an issue if there is workflow which will be checkpointed,
and another persisted workflow references some other entity as a parent workflow
---
.../brooklyn/api/mgmt/ManagementContext.java | 9 ++--
.../camp/brooklyn/SshCommandSensorYamlTest.java | 19 +++++---
.../mgmt/internal/EntityManagementSupport.java | 34 ++++++++-----
.../core/mgmt/internal/LocalManagementContext.java | 3 +-
.../util/core/task/BasicExecutionContext.java | 55 +++++++++++++++++-----
.../brooklyn/core/mgmt/ha/HotStandbyTest.java | 2 +-
.../system_service/SystemServiceEnricher.java | 3 +-
7 files changed, 86 insertions(+), 39 deletions(-)
diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/ManagementContext.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/ManagementContext.java
index 356533e089..8052af3961 100644
--- a/api/src/main/java/org/apache/brooklyn/api/mgmt/ManagementContext.java
+++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/ManagementContext.java
@@ -156,12 +156,9 @@ public interface ManagementContext {
*/
SubscriptionManager getSubscriptionManager();
- //TODO (Alex) I'm not sure the following two getXxxContext methods are needed on the interface;
- //I expect they will only be called once, in AbstractEntity, and fully capable
- //there of generating the respective contexts from the managers
- //(Litmus test will be whether there is anything in FederatedManagementContext
- //which requires a custom FederatedExecutionContext -- or whether BasicEC
- //works with FederatedExecutionManager)
+ // not sure the following two getXxxContext methods are desired on this interface;
+ // almost everyone should use the entity.getExecutionContext() which is a shared instance with better blocking;
+ // this should just be called to initialize and for special cases where we want to bypass that startup blocking
/**
* Returns an {@link ExecutionContext} instance representing tasks
* (from the {@link ExecutionManager}) associated with this entity, and capable
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/SshCommandSensorYamlTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/SshCommandSensorYamlTest.java
index 644909b555..9bb4efa49c 100644
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/SshCommandSensorYamlTest.java
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/SshCommandSensorYamlTest.java
@@ -31,6 +31,7 @@ import org.apache.brooklyn.core.entity.RecordingSensorEventListener;
import org.apache.brooklyn.core.sensor.Sensors;
import org.apache.brooklyn.core.test.entity.TestApplication;
import org.apache.brooklyn.entity.software.base.VanillaSoftwareProcess;
+import org.apache.brooklyn.entity.stock.BasicApplication;
import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.core.internal.ssh.RecordingSshTool;
@@ -38,6 +39,7 @@ import org.apache.brooklyn.util.core.internal.ssh.RecordingSshTool.CustomRespons
import org.apache.brooklyn.util.core.internal.ssh.RecordingSshTool.ExecParams;
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.BeforeMethod;
@@ -232,10 +234,9 @@ public class SshCommandSensorYamlTest extends AbstractYamlTest {
@Test(groups="Integration") // because slow
public void testSshCommandSensorPeriodicFeedServiceUpFalseDoesNotRunAtStartup() throws Exception {
- RecordingSshTool.setCustomResponse(".*myCommand.*", new RecordingSshTool.CustomResponse(0, "myResponse", null));
+ RecordingSshTool.setCustomResponse(".*myCommand.*", new RecordingSshTool.CustomResponse(0, "myResponse0", null));
- Stopwatch sw = Stopwatch.createStarted();
- Entity app = createAndStartApplication(
+ BasicApplication app = (BasicApplication) createAndStartApplication(
"location:",
" localhost:",
" sshToolClass: "+RecordingSshTool.class.getName(),
@@ -248,13 +249,19 @@ public class SshCommandSensorYamlTest extends AbstractYamlTest {
" brooklyn.config:",
" name: mySensor",
" command: myCommand",
- " period: 5s",
+ " period: 2s",
" onlyIfServiceUp: true");
waitForApplicationTasks(app);
VanillaSoftwareProcess entity = (VanillaSoftwareProcess) Iterables.getOnlyElement(app.getChildren());
- EntityAsserts.assertAttributeEqualsEventually(entity, Sensors.newStringSensor("mySensor"), "myResponse");
- Asserts.assertThat(Duration.of(sw), d -> d.isLongerThan(Duration.seconds(4)));
+ EntityAsserts.assertAttributeEqualsEventually(entity, Sensors.newStringSensor("mySensor"), "myResponse0");
+
+ // once run once, it shouldn't run again for 2s (plus or minus 1s tolerance here)
+ Stopwatch sw = Stopwatch.createStarted();
+ RecordingSshTool.setCustomResponse(".*myCommand.*", new RecordingSshTool.CustomResponse(0, "myResponse1", null));
+ EntityAsserts.assertAttributeEqualsEventually(entity, Sensors.newStringSensor("mySensor"), "myResponse1");
+ Asserts.assertThat(Duration.of(sw), d -> d.isLongerThan(Duration.seconds(1)));
+ Asserts.assertThat(Duration.of(sw), d -> d.isShorterThan(Duration.seconds(3)));
}
@Test
diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/EntityManagementSupport.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/EntityManagementSupport.java
index c3ea682c3e..366b9ab05d 100644
--- a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/EntityManagementSupport.java
+++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/EntityManagementSupport.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import org.apache.brooklyn.api.effector.Effector;
@@ -52,6 +53,7 @@ import org.apache.brooklyn.core.objs.AbstractEntityAdjunct;
import org.apache.brooklyn.core.workflow.DanglingWorkflowException;
import org.apache.brooklyn.core.workflow.WorkflowExecutionContext;
import org.apache.brooklyn.core.workflow.store.WorkflowStatePersistenceViaSensors;
+import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.core.task.DynamicTasks;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.exceptions.Exceptions;
@@ -231,10 +233,11 @@ public class EntityManagementSupport {
this.managementContext = info.getManagementContext();
nonDeploymentManagementContext.setMode(NonDeploymentManagementContextMode.MANAGEMENT_STARTING);
- if (!isReadOnly()) {
- nonDeploymentManagementContext.getSubscriptionManager().setDelegate((AbstractSubscriptionManager) managementContext.getSubscriptionManager());
- nonDeploymentManagementContext.getSubscriptionManager().startDelegatingForSubscribing();
- }
+ // defer this until mgmt context started, so all other entities will be known, in case they are accessed in the tasks
+// if (!isReadOnly()) {
+// nonDeploymentManagementContext.getSubscriptionManager().setDelegate((AbstractSubscriptionManager) managementContext.getSubscriptionManager());
+// nonDeploymentManagementContext.getSubscriptionManager().startDelegatingForSubscribing();
+// }
managementContextUsable.set(true);
currentlyDeployed.set(true);
@@ -280,9 +283,13 @@ public class EntityManagementSupport {
entity.onManagementStarting();
// start those policies etc which are labelled as auto-start
- entity.policies().forEach(adj -> { if (adj instanceof EntityAdjunct.AutoStartEntityAdjunct) ((EntityAdjunct.AutoStartEntityAdjunct)adj).start(); });
- entity.enrichers().forEach(adj -> { if (adj instanceof EntityAdjunct.AutoStartEntityAdjunct) ((EntityAdjunct.AutoStartEntityAdjunct)adj).start(); });
- entity.feeds().forEach(f -> { if (!f.isActivated()) f.start(); });
+ BiConsumer<String,Runnable> queueTask = (name, r) -> entity.getExecutionContext().submit(name, r);
+ entity.policies().forEach(adj -> { if (adj instanceof EntityAdjunct.AutoStartEntityAdjunct)
+ queueTask.accept("Start policy "+adj, ((EntityAdjunct.AutoStartEntityAdjunct)adj)::start); });
+ entity.enrichers().forEach(adj -> { if (adj instanceof EntityAdjunct.AutoStartEntityAdjunct)
+ queueTask.accept("Start enricher "+adj, ((EntityAdjunct.AutoStartEntityAdjunct)adj)::start); });
+ entity.feeds().forEach(f -> { if (!f.isActivated())
+ queueTask.accept("Start feed "+f, f::start); });
if (AUTO_FAIL_AND_RESUME_WORKFLOWS) {
// resume any workflows that were dangling due to shutdown
@@ -293,7 +300,7 @@ public class EntityManagementSupport {
.collect(Collectors.toList());
if (!shutdownInterruptedWorkflows.isEmpty()) {
log.debug("Discovered workflows noted as 'interrupted' on startup at "+entity+", will resume as dangling: "+shutdownInterruptedWorkflows);
- entity.getExecutionContext().submit(DynamicTasks.of("Resuming with failure " + shutdownInterruptedWorkflows.size() + " interrupted workflow" + (shutdownInterruptedWorkflows.size() != 1 ? "s" : ""), () -> {
+ getManagementContext().getExecutionContext(entity).submit(DynamicTasks.of("Resuming with failure " + shutdownInterruptedWorkflows.size() + " interrupted workflow" + (shutdownInterruptedWorkflows.size() != 1 ? "s" : ""), () -> {
shutdownInterruptedWorkflows.forEach(w -> {
// these are backgrounded because they are expected to fail
// we also have to wait until mgmt is complete
@@ -346,9 +353,7 @@ public class EntityManagementSupport {
/* on start, we want to:
* - set derived/inherited config values (not needed, the specs should have taken care of that?)
* - publish all queued sensors (done below)
- * - start all queued executions
- * (e.g. subscription delivery - done below? are there others and if so how are they unlocked?
- * curious where the "start queued tasks" logic is; must be somewhere as it all seems to have been working fine (Aug 2016))
+ * - start all queued executions (unpause entity's execution context, subscription delivery)
* [in exactly this order, at each entity]
* then subsequent sensor events and executions occur directly (no queueing)
*
@@ -357,7 +362,10 @@ public class EntityManagementSupport {
*/
if (!isReadOnly()) {
+ nonDeploymentManagementContext.getSubscriptionManager().setDelegate((AbstractSubscriptionManager) managementContext.getSubscriptionManager());
nonDeploymentManagementContext.getSubscriptionManager().startDelegatingForPublishing();
+ nonDeploymentManagementContext.getSubscriptionManager().startDelegatingForSubscribing();
+ ((BasicExecutionContext)getExecutionContext()).unpause();
}
if (!isReadOnly()) {
@@ -520,7 +528,9 @@ public class EntityManagementSupport {
if (managementContextUsable.get()) {
synchronized (this) {
if (executionContext!=null) return executionContext;
- executionContext = managementContext.getExecutionContext(entity);
+ ExecutionContext newExecutionContext = managementContext.getExecutionContext(entity);
+ ((BasicExecutionContext)newExecutionContext).pause(); // start paused, so things don't run until mgmt is started, and all entities known
+ executionContext = newExecutionContext;
return executionContext;
}
}
diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalManagementContext.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalManagementContext.java
index db284614f1..4fb32e5230 100644
--- a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalManagementContext.java
+++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalManagementContext.java
@@ -47,6 +47,7 @@ import org.apache.brooklyn.api.mgmt.TaskAdaptable;
import org.apache.brooklyn.api.mgmt.ha.ManagementNodeState;
import org.apache.brooklyn.core.BrooklynFeatureEnablement;
import org.apache.brooklyn.core.effector.Effectors;
+import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.entity.drivers.downloads.BasicDownloadsManager;
import org.apache.brooklyn.core.internal.BrooklynInitialization;
import org.apache.brooklyn.core.internal.BrooklynProperties;
@@ -360,7 +361,7 @@ public class LocalManagementContext extends AbstractManagementContext {
}
protected <T> Task<T> runAtEntity(Entity entity, TaskAdaptable<T> task) {
- getExecutionContext(entity).submit(task);
+ ((EntityInternal)entity).getExecutionContext().submit(task);
if (DynamicTasks.getTaskQueuingContext()!=null) {
// put it in the queueing context so it appears in the GUI
// mark it inessential as this is being invoked from code,
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index dc5dfad97f..b0757bacc8 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -18,10 +18,6 @@
*/
package org.apache.brooklyn.util.core.task;
-import com.google.common.annotations.Beta;
-import com.google.common.base.Function;
-import com.google.common.base.Supplier;
-import com.google.common.collect.Iterables;
import java.lang.reflect.Proxy;
import java.util.ArrayDeque;
import java.util.ArrayList;
@@ -29,6 +25,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
@@ -39,6 +36,11 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+
+import com.google.common.annotations.Beta;
+import com.google.common.base.Function;
+import com.google.common.base.Supplier;
+import com.google.common.collect.Iterables;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ExecutionContext;
import org.apache.brooklyn.api.mgmt.ExecutionManager;
@@ -51,6 +53,7 @@ import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags.WrappedEntity;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags.WrappedItem;
import org.apache.brooklyn.core.mgmt.entitlement.Entitlements;
+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.task.BasicExecutionManager.BrooklynTaskLoggingMdc;
@@ -60,6 +63,7 @@ import org.apache.brooklyn.util.guava.Maybe;
import org.apache.brooklyn.util.javalang.Threads;
import org.apache.brooklyn.util.time.CountdownTimer;
import org.apache.brooklyn.util.time.Duration;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -404,17 +408,44 @@ public class BasicExecutionContext extends AbstractExecutionContext {
});
}
- if (task instanceof Task) {
- return executionManager.submit(properties, (Task)task);
- } else if (task instanceof Callable) {
- return executionManager.submit(properties, (Callable)task);
- } else if (task instanceof Runnable) {
- return (Task<T>) executionManager.submit(properties, (Runnable)task);
- } else {
- throw new IllegalArgumentException("Unhandled task type: task="+task+"; type="+(task!=null ? task.getClass() : "null"));
+ return submitViaExecutionManagerOrHold(task, properties);
+ }
+
+ boolean paused = false;
+ List<Pair<Task,Map>> tasksQueuedWhilePaused = MutableList.of();
+ public void pause() {
+ this.paused = true;
+ }
+ public void unpause() {
+ synchronized (tasksQueuedWhilePaused) {
+ tasksQueuedWhilePaused.forEach(pair -> submitWithoutCheckingPaused(pair.getLeft(), pair.getRight()));
+ tasksQueuedWhilePaused.clear();
+ this.paused = false;
}
}
+ private <T> Task submitViaExecutionManagerOrHold(Object task, Map properties) {
+ Task taskT = null;
+ if (task instanceof Task) taskT = (Task) task;
+ else if (task instanceof TaskAdaptable) taskT = ((TaskAdaptable) task).asTask();
+ else if (task instanceof Callable) taskT = new BasicTask(properties, (Callable)task);
+ else if (task instanceof Runnable) taskT = new BasicTask(properties, (Runnable)task);
+ else throw new IllegalArgumentException("Unhandled task type: task="+ task +"; type="+(task !=null ? task.getClass() : "null"));
+
+ if (paused) {
+ synchronized (tasksQueuedWhilePaused) {
+ if (paused) {
+ tasksQueuedWhilePaused.add(Pair.of(taskT, properties));
+ return taskT;
+ }
+ }
+ }
+ return submitWithoutCheckingPaused(taskT, properties);
+ }
+ private <T> Task submitWithoutCheckingPaused(Task task, Map properties) {
+ return executionManager.submit(properties, task);
+ }
+
private String idStack(Entity target) {
Deque<String> ids = new ArrayDeque<>();
Entity e = target;
diff --git a/core/src/test/java/org/apache/brooklyn/core/mgmt/ha/HotStandbyTest.java b/core/src/test/java/org/apache/brooklyn/core/mgmt/ha/HotStandbyTest.java
index f1e7a41b0b..87570bc983 100644
--- a/core/src/test/java/org/apache/brooklyn/core/mgmt/ha/HotStandbyTest.java
+++ b/core/src/test/java/org/apache/brooklyn/core/mgmt/ha/HotStandbyTest.java
@@ -626,7 +626,7 @@ public class HotStandbyTest {
forcePersistNow(n1);
Assert.assertTrue(entity.feeds().getFeeds().size() > 0, "Feeds: "+entity.feeds().getFeeds());
for (Feed feed : entity.feeds().getFeeds()) {
- assertTrue(feed.isRunning(), "Feed expected running, but it is non-running");
+ Asserts.eventually(() -> feed, Feed::isRunning, Duration.seconds(2), Duration.millis(10), "Feed expected running, but it is non-running");
}
HaMgmtNode n2 = createHotStandby(Duration.PRACTICALLY_FOREVER);
diff --git a/software/base/src/main/java/org/apache/brooklyn/entity/system_service/SystemServiceEnricher.java b/software/base/src/main/java/org/apache/brooklyn/entity/system_service/SystemServiceEnricher.java
index ac9b4122ae..cfcf407460 100644
--- a/software/base/src/main/java/org/apache/brooklyn/entity/system_service/SystemServiceEnricher.java
+++ b/software/base/src/main/java/org/apache/brooklyn/entity/system_service/SystemServiceEnricher.java
@@ -29,6 +29,7 @@ import org.apache.brooklyn.core.config.ConfigKeys;
import org.apache.brooklyn.core.effector.EffectorTasks;
import org.apache.brooklyn.core.enricher.AbstractEnricher;
import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags.WrappedStream;
import org.apache.brooklyn.entity.software.base.SoftwareProcess;
@@ -128,6 +129,6 @@ public class SystemServiceEnricher extends AbstractEnricher implements Enricher
}
ExecutionContext getEntityExecutionContext() {
- return getManagementContext().getExecutionContext(entity);
+ return ((EntityInternal)entity).getExecutionContext();
}
}