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 2025/01/22 23:28:02 UTC
(brooklyn-server) 04/04: improve wait and recalculation for in-flight racing updates to service up, problems, and expected
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 665dc5331a7ce9b9e92087dd3ee2b9e75e5acddc
Author: Alex Heneveld <gi...@alex.heneveld.org>
AuthorDate: Wed Jan 22 21:14:32 2025 +0000
improve wait and recalculation for in-flight racing updates to service up, problems, and expected
setExpected will wait a while, because that is called explicitly when children tasks have finished
(and should have concluded their expected) but might not have delivered all their sensors up to the parent;
it stops waiting if there are no unrelated subscription deliveries in flight
---
.../core/entity/lifecycle/ServiceStateLogic.java | 225 ++++++++++++---------
.../blueprints/ChildrenQuorateRaceTest.java | 205 +++++++++++++++++++
2 files changed, 340 insertions(+), 90 deletions(-)
diff --git a/core/src/main/java/org/apache/brooklyn/core/entity/lifecycle/ServiceStateLogic.java b/core/src/main/java/org/apache/brooklyn/core/entity/lifecycle/ServiceStateLogic.java
index 9310a44b99..baf7c644bc 100644
--- a/core/src/main/java/org/apache/brooklyn/core/entity/lifecycle/ServiceStateLogic.java
+++ b/core/src/main/java/org/apache/brooklyn/core/entity/lifecycle/ServiceStateLogic.java
@@ -34,6 +34,7 @@ import org.apache.brooklyn.api.effector.Effector;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.entity.EntityLocal;
import org.apache.brooklyn.api.entity.Group;
+import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.api.sensor.Enricher;
import org.apache.brooklyn.api.sensor.EnricherSpec;
@@ -51,9 +52,9 @@ import org.apache.brooklyn.core.entity.Attributes;
import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.EntityAdjuncts;
import org.apache.brooklyn.core.entity.EntityInternal;
-import org.apache.brooklyn.core.entity.EntityPredicates;
import org.apache.brooklyn.core.entity.lifecycle.Lifecycle.Transition;
import org.apache.brooklyn.core.entity.trait.Startable;
+import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.sensor.BasicSensorEvent;
import org.apache.brooklyn.enricher.stock.AbstractMultipleSensorAggregator;
import org.apache.brooklyn.enricher.stock.Enrichers;
@@ -64,6 +65,7 @@ import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.collections.QuorumCheck;
import org.apache.brooklyn.util.collections.QuorumCheck.QuorumChecks;
+import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.core.task.ValueResolver;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.guava.Functionals;
@@ -164,8 +166,8 @@ public class ServiceStateLogic {
}
private static void setExpectedState(Entity entity, Lifecycle state, boolean waitBrieflyForServiceUpIfRunning) {
- if (waitBrieflyForServiceUpIfRunning) {
- recomputeIfIssueWhenBecomingExpectedRunning("setting expected state", entity, state);
+ if (waitBrieflyForServiceUpIfRunning && state == Lifecycle.RUNNING) {
+ recomputeWantingNoIssuesWhenBecomingExpectedRunning("setting expected state", entity, RecomputeWaitMode.LONG);
}
((EntityInternal)entity).sensors().set(Attributes.SERVICE_STATE_EXPECTED, new Lifecycle.Transition(state, new Date()));
@@ -181,67 +183,97 @@ public class ServiceStateLogic {
return expected.getState();
}
- public static void recomputeIfIssueWhenBecomingExpectedRunning(String when, Entity entity, Lifecycle state) {
- if (!Entities.isManagedActive(entity) || state!=Lifecycle.RUNNING) {
- return;
- } else {
- Map<String, Object> problems = entity.getAttribute(SERVICE_PROBLEMS);
- boolean noProblems = problems == null || problems.isEmpty();
- Boolean up = entity.getAttribute(Attributes.SERVICE_UP);
- if (Boolean.TRUE.equals(up) && noProblems) {
- return;
- } else {
- log.debug("Service not up pre-check, up="+up+" and problems="+problems+" when setting "+ state +" (when "+when+") on " + entity+"; possibly just needs a recompute; doing recompute now");
-
- try {
- Iterables.filter(entity.enrichers(), x -> x instanceof ComputeServiceIndicatorsFromChildrenAndMembers).forEach(
- x -> {
- ComputeServiceIndicatorsFromChildrenAndMembers mx = (ComputeServiceIndicatorsFromChildrenAndMembers) x;
- if (mx.isRunning()) {
- log.debug("Service not up pre-check recompute rerunning "+mx);
- mx.onUpdated();
- }
- }
- );
-
- Map<String, Object> notUpIndicators = entity.sensors().get(Attributes.SERVICE_NOT_UP_INDICATORS);
- if (notUpIndicators == null || notUpIndicators.isEmpty()) {
- Maybe<Enricher> css = EntityAdjuncts.tryFindWithUniqueTag(entity.enrichers(), ServiceNotUpLogic.DEFAULT_ENRICHER_UNIQUE_TAG);
- if (css.isPresent()) {
- SensorEvent<Map<String, Object>> pseudoEvent = new BasicSensorEvent<>(Attributes.SERVICE_NOT_UP_INDICATORS, entity, notUpIndicators);
- ((SensorEventListener) css.get()).onEvent(pseudoEvent);
- up = entity.getAttribute(Attributes.SERVICE_UP);
- log.debug("Service not up pre-check recompute ran, service.isUp="+up+" after: "+css);
- }
- } else {
- log.debug("Service not up pre-check recompute not running because not up indicators are now: " + notUpIndicators);
- }
- } catch (Exception e) {
- Exceptions.propagateIfFatal(e);
- log.debug("Service is not up when setting "+ state +" on " + entity+", and attempt to run standard prep workflows failed with exception; will wait and see if service up clears, but for reference the error is: "+e);
- if (log.isTraceEnabled()) log.trace("Exception trace", e);
- }
-
- if (!Boolean.TRUE.equals(up) && Entities.isManagedActive(entity)) {
- // pause briefly to allow any recent problem-clearing processing to complete;
- // should be less necessary now that the code above explicitly triggers any not-up indicators
- Stopwatch timer = Stopwatch.createStarted();
- boolean nowUp = Repeater.create()
- .every(ValueResolver.REAL_QUICK_PERIOD)
- .limitTimeTo(ValueResolver.PRETTY_QUICK_WAIT)
- .until(entity, EntityPredicates.attributeEqualTo(Attributes.SERVICE_UP, true))
- .run();
- if (nowUp) {
- log.debug("Had to wait " + Duration.of(timer) + " for " + entity + " " + Attributes.SERVICE_UP + " to be true before setting " + state);
- } else {
- if (Entities.isManagedActive(entity)) {
- log.warn("Service is not up when "+when+" on " + entity + "; delayed " + Duration.of(timer) + " "
- + "but " + Attributes.SERVICE_UP + " did not recover from " + up + "; not-up-indicators=" + entity.getAttribute(Attributes.SERVICE_NOT_UP_INDICATORS));
+ private static boolean isEmptyOrNull(Map x) {
+ return x==null || x.isEmpty();
+ }
+ private enum RecomputeWaitMode { LONG, SHORT, RECOMPUTE_ONLY, NONE }
+ private static boolean recomputeWantingNoIssuesWhenBecomingExpectedRunning(String when, Entity entity, RecomputeWaitMode canWait) {
+ if (!Entities.isManagedActive(entity)) return true;
+
+ Map<String, Object> problems = entity.getAttribute(SERVICE_PROBLEMS);
+ Boolean up = entity.getAttribute(Attributes.SERVICE_UP);
+ if (Boolean.TRUE.equals(up) && isEmptyOrNull(problems)) return true;
+ if (canWait==RecomputeWaitMode.NONE) return false;
+
+ log.debug("Recompute indicated setting RUNNING ("+when+") on service issue; up="+up+", problems="+problems+",on " + entity + " (mode "+canWait+")");
+ try {
+ Iterables.filter(entity.enrichers(), x -> x instanceof ComputeServiceIndicatorsFromChildrenAndMembers).forEach(
+ x -> {
+ ComputeServiceIndicatorsFromChildrenAndMembers mx = (ComputeServiceIndicatorsFromChildrenAndMembers) x;
+ if (mx.isRunning()) {
+ log.debug("Recompute rerunning "+mx);
+ mx.onUpdated();
+ log.debug("Recomputed values now: problems="+
+ entity.sensors().get(SERVICE_PROBLEMS) + ", not_up_indicators=" + entity.sensors().get(SERVICE_NOT_UP_INDICATORS) );
}
}
+ );
+
+ Map<String, Object> notUpIndicators = entity.sensors().get(Attributes.SERVICE_NOT_UP_INDICATORS);
+ if (notUpIndicators == null || notUpIndicators.isEmpty()) {
+ Maybe<Enricher> css = EntityAdjuncts.tryFindWithUniqueTag(entity.enrichers(), ServiceNotUpLogic.DEFAULT_ENRICHER_UNIQUE_TAG);
+ if (css.isPresent()) {
+ SensorEvent<Map<String, Object>> pseudoEvent = new BasicSensorEvent<>(Attributes.SERVICE_NOT_UP_INDICATORS, entity, notUpIndicators);
+ ((SensorEventListener) css.get()).onEvent(pseudoEvent);
+ up = entity.getAttribute(Attributes.SERVICE_UP);
+ log.debug("Recompute for service indicators now gives: service.isUp="+up+" after: "+css);
}
+ } else {
+ log.debug("Recomputed not_up_indicators now: " + notUpIndicators);
}
- }
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ if (!Entities.isManagedActive(entity)) return true;
+ log.debug("Service is not up when setting RUNNING on " + entity+", and attempt to run standard prep workflows failed with exception: "+e);
+ if (log.isTraceEnabled()) log.trace("Exception trace", e);
+ }
+
+ if (recomputeWantingNoIssuesWhenBecomingExpectedRunning(when+" (after recompute)", entity, RecomputeWaitMode.NONE)) return true;
+ if (canWait==RecomputeWaitMode.RECOMPUTE_ONLY) return false;
+
+ assert canWait==RecomputeWaitMode.LONG || canWait==RecomputeWaitMode.SHORT;
+ // repeat with pauses briefly to allow any recent in-flight service-ups or state changes or problem-clearing processing to complete;
+ // should only be necessary when using setExpectedState, where canWait is true, unless the actual is used without expected;
+ // but leaving it as it used to be to minimise surprises for now;
+ // probably also no need to recompute above either with this (could be tidied up a lot)
+ Stopwatch timer = Stopwatch.createStarted();
+ final Duration LONG_WAIT = Duration.THIRTY_SECONDS; // this should be enough time for in-flight racing activities, even when slow and heavily contended
+ Task current = Tasks.current();
+ boolean notTimedout = Repeater.create()
+ .every(ValueResolver.REAL_QUICK_PERIOD)
+ .limitTimeTo(canWait==RecomputeWaitMode.LONG ? LONG_WAIT : ValueResolver.PRETTY_QUICK_WAIT)
+ .until(() -> {
+ Set<Task<?>> tasksHere = BrooklynTaskTags.getTasksInEntityContext(
+ ((EntityInternal) entity).getManagementContext().getExecutionManager(), entity);
+ java.util.Optional<Task<?>> unrelatedSubmission = tasksHere.stream()
+ .filter(t ->
+ !t.isDone() &&
+ BrooklynTaskTags.hasTag(t, BrooklynTaskTags.SENSOR_TAG) &&
+ !Tasks.isAncestor(current, anc -> Objects.equal(anc, t)))
+ .findAny();
+ // abort when self and children have no unrelated submission tasks, or if something is known to be on_fire;
+ // otherwise give it LONG_WAIT (arbitrary but inoffensive) to settle in that case (things running very slow)
+ if (!unrelatedSubmission.isPresent()) return true;
+ return recomputeWantingNoIssuesWhenBecomingExpectedRunning(when+" (recheck after "+Duration.of(timer.elapsed())+")", entity, RecomputeWaitMode.NONE);
+ })
+ .run();
+
+ boolean nowUp = recomputeWantingNoIssuesWhenBecomingExpectedRunning(when+" (recheck after "+
+ (notTimedout ? "completion" : "timeout")+", after "+Duration.of(timer.elapsed())+")", entity, RecomputeWaitMode.RECOMPUTE_ONLY);
+ if (nowUp) {
+ log.debug("Recompute determined "+entity+" is up, after " + Duration.of(timer));
+ return true;
+ }
+
+ if (!Entities.isManagedActive(entity)) return true;
+
+ log.warn("Service is not up when "+when+" on " + entity + "; delayed " + Duration.of(timer) + " "
+ + "but: " + Attributes.SERVICE_UP + "=" + entity.getAttribute(Attributes.SERVICE_UP) + ", "
+ + "not-up-indicators=" + entity.getAttribute(Attributes.SERVICE_NOT_UP_INDICATORS)
+ + ", problems=" + entity.getAttribute(Attributes.SERVICE_PROBLEMS)
+ );
+ // slight chance above has updated since the check, but previuos log messages should make clear what happened
+ return false;
}
public static Lifecycle getActualState(Entity entity) {
@@ -309,7 +341,7 @@ public class ServiceStateLogic {
.defaultValue(Entities.UNCHANGED).apply(input.getValue());
if (!Objects.equal(result, Entities.UNCHANGED)) {
Boolean prevValue = entity.sensors().get(SERVICE_UP);
- if (!Objects.equal(result, prevValue)) {
+ if (!Objects.equal(result, prevValue) && (prevValue!=null || result!=Boolean.TRUE)) {
log.debug("Enricher '" + DEFAULT_ENRICHER_UNIQUE_TAG + "' for " + entity + " determined service up changed from " + prevValue + " to " + result + " due to indicators: " + input);
}
}
@@ -371,39 +403,52 @@ public class ServiceStateLogic {
}
}
+ transient int recomputeDepth=0;
protected Maybe<Lifecycle> computeActualStateWhenExpectedRunning(SensorEvent<Object> event) {
- int count=0;
- while (true) {
- Map<String, Object> problems = entity.getAttribute(SERVICE_PROBLEMS);
- boolean noProblems = problems == null || problems.isEmpty();
+ if (recomputeDepth>0) {
+ return Maybe.absent("Skipping actual state computation because already computing");
+ }
+ try {
+ while (true) {
+ Map<String, Object> problems = entity.getAttribute(SERVICE_PROBLEMS);
+ boolean noProblems = problems == null || problems.isEmpty();
- Boolean serviceUp = entity.getAttribute(SERVICE_UP);
+ Boolean serviceUp = entity.getAttribute(SERVICE_UP);
- if (Boolean.TRUE.equals(serviceUp) && noProblems) {
- return Maybe.of(Lifecycle.RUNNING);
- } else {
- if (!Entities.isManagedActive(entity)) {
- return Maybe.absent("entity not managed active");
- }
- if (!Lifecycle.ON_FIRE.equals(entity.getAttribute(SERVICE_STATE_ACTUAL))) {
- boolean waitable = count==0;
- waitable = waitable && event!=null && !Attributes.SERVICE_UP.equals(event.getSensor());
- if (waitable) {
- // very occasional race here; might want to give a grace period if entity has just transitioned; allow children to catch up
- // we probably did the wait when expected running, but possibly in some cases we don't (seen once, 2024-07, not reproduced)
- log.debug("Entity "+entity+" would be on-fire due to problems (up="+serviceUp+", problems="+problems+"), will attempt re-check");
- recomputeIfIssueWhenBecomingExpectedRunning("computing actual state", entity, Lifecycle.RUNNING);
- count++;
- continue;
+ if (Boolean.TRUE.equals(serviceUp) && noProblems) {
+ return Maybe.of(Lifecycle.RUNNING);
+ } else {
+ if (!Entities.isManagedActive(entity)) {
+ return Maybe.absent("entity not managed active");
+ }
+ // with delay when writing expected state, it should not be necessary to have a wait/retry
+ if (!Lifecycle.ON_FIRE.equals(entity.getAttribute(SERVICE_STATE_ACTUAL))) {
+ boolean retryable = recomputeDepth == 0;
+ // allow recompute if event is null (intermediate recomputation?)
+ // but need to prevent
+ retryable = retryable && (event == null || !Attributes.SERVICE_UP.equals(event.getSensor()));
+ if (retryable) {
+ recomputeDepth++;
+ // occasional race here; might want to give a grace period if entity has just transitioned; allow children to catch up;
+ // we should have done the wait when expected running, but possibly it hasn't caught up yet
+ log.debug("Entity " + entity + " would be computed on-fire due to problems (up=" + serviceUp + ", problems=" + problems + "), will attempt re-check");
+ recomputeWantingNoIssuesWhenBecomingExpectedRunning("computing actual state", entity,
+ RecomputeWaitMode.SHORT // NONE would probalby be fine here, with none of the recompute above,
+ // at least whenever expected state is used, due to how it waits now; but leaving it as is until more confirmation
+ );
+ continue;
+ }
}
+ BrooklynLogging.log(log, BrooklynLogging.levelDependingIfReadOnly(entity, LoggingLevel.WARN, LoggingLevel.TRACE, LoggingLevel.DEBUG),
+ "Setting " + entity + " " + Lifecycle.ON_FIRE + " due to problems when expected running, " +
+ "trigger=" + event + ", " +
+ "up=" + serviceUp + ", " +
+ (noProblems ? "not-up-indicators: " + entity.getAttribute(SERVICE_NOT_UP_INDICATORS) : "problems: " + problems));
+ return Maybe.of(Lifecycle.ON_FIRE);
}
- BrooklynLogging.log(log, BrooklynLogging.levelDependingIfReadOnly(entity, LoggingLevel.WARN, LoggingLevel.TRACE, LoggingLevel.DEBUG),
- "Setting " + entity + " " + Lifecycle.ON_FIRE + " due to problems when expected running, " +
- "trigger="+event+", "+
- "up=" + serviceUp + ", " +
- (noProblems ? "not-up-indicators: " + entity.getAttribute(SERVICE_NOT_UP_INDICATORS) : "problems: " + problems));
- return Maybe.of(Lifecycle.ON_FIRE);
}
+ } finally {
+ recomputeDepth = 0;
}
}
@@ -421,7 +466,7 @@ public class ServiceStateLogic {
return Maybe.of(Lifecycle.STOPPED);
} else {
BrooklynLogging.log(log, BrooklynLogging.levelDependingIfReadOnly(entity, LoggingLevel.WARN, LoggingLevel.TRACE, LoggingLevel.DEBUG),
- "Setting "+entity+" "+Lifecycle.ON_FIRE+" due to problems when expected "+stateTransition+" / up="+up+": "+problems);
+ "Computed "+entity+" "+Lifecycle.ON_FIRE+" due to problems when expected "+stateTransition+" / up="+up+": "+problems);
return Maybe.of(Lifecycle.ON_FIRE);
}
} else {
@@ -549,7 +594,7 @@ public class ServiceStateLogic {
@SuppressWarnings("serial")
public static final ConfigKey<Set<Lifecycle>> IGNORE_ENTITIES_WITH_THESE_SERVICE_STATES = ConfigKeys.newConfigKey(new TypeToken<Set<Lifecycle>>() {},
"enricher.service_state.children_and_members.ignore_entities.service_state_values",
- "Service states (including null) which indicate an entity should be ignored when looking at children service states; anything apart from RUNNING not in this list will be treated as not healthy (by default just ON_FIRE will mean not healthy)",
+ "Service states of children (including null) which indicate they should be ignored when looking at children service states; anything apart from RUNNING not in this list will be treated as not healthy (by default just ON_FIRE will mean not healthy)",
MutableSet.<Lifecycle>builder().addAll(Lifecycle.values()).add(null).remove(Lifecycle.RUNNING).remove(Lifecycle.ON_FIRE).build().asUnmodifiable());
protected String getKeyForMapSensor() {
diff --git a/launcher/src/test/java/org/apache/brooklyn/launcher/blueprints/ChildrenQuorateRaceTest.java b/launcher/src/test/java/org/apache/brooklyn/launcher/blueprints/ChildrenQuorateRaceTest.java
new file mode 100644
index 0000000000..59e5dcf0a6
--- /dev/null
+++ b/launcher/src/test/java/org/apache/brooklyn/launcher/blueprints/ChildrenQuorateRaceTest.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.brooklyn.launcher.blueprints;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.EntitySpec;
+import org.apache.brooklyn.api.mgmt.ManagementContext;
+import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.config.ConfigKey;
+import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.Dumper;
+import org.apache.brooklyn.core.entity.EntityAsserts;
+import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
+import org.apache.brooklyn.core.mgmt.EntityManagementUtils;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.core.workflow.WorkflowBasicTest;
+import org.apache.brooklyn.entity.stock.BasicApplication;
+import org.apache.brooklyn.entity.stock.BasicStartable;
+import org.apache.brooklyn.entity.stock.WorkflowStartable;
+import org.apache.brooklyn.util.collections.MutableList;
+import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.collections.QuorumCheck;
+import org.apache.brooklyn.util.core.task.BasicExecutionManager;
+import org.apache.brooklyn.util.core.task.Tasks;
+import org.apache.brooklyn.util.time.Duration;
+import org.apache.brooklyn.util.time.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.Test;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import static org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ComputeServiceIndicatorsFromChildrenAndMembers.UP_QUORUM_CHECK;
+
+/** This does rebind. See SimpleBlueprintNonRebindTest for an example with rebind disabled. */
+public class ChildrenQuorateRaceTest extends AbstractBlueprintTest {
+
+ private static final Logger log = LoggerFactory.getLogger(ChildrenQuorateRaceTest.class);
+
+ @Override
+ protected ManagementContext decorateManagementContext(ManagementContext mgmt) {
+ mgmt = super.decorateManagementContext(mgmt);
+ // make workflow step types available
+ WorkflowBasicTest.addWorkflowStepTypes(mgmt);
+ return mgmt;
+ }
+
+ @Override
+ protected boolean isViewerEnabled() {
+ return true;
+ }
+
+ @Override
+ protected boolean isUsingNewViewerForRebind() {
+ return true;
+ }
+
+ public Duration randomNormalishJitter(Duration base, Duration jitterMax) {
+ boolean increase = base.isLongerThan(Duration.ZERO) ? Math.random() > 0.5 : true;
+ if (!increase && jitterMax.isLongerThan(base)) jitterMax = base;
+ Duration jitterActual = jitterMax.multiply(Math.random() * Math.random());
+ if (!increase) jitterActual = jitterActual.multiply(-1);
+ jitterActual = Duration.millis(jitterActual.toMilliseconds());
+ return base.add(jitterActual);
+ }
+ public Duration randomNormalishJitter(String base, String jitterMax) {
+ return randomNormalishJitter(Duration.of(base), Duration.of(jitterMax));
+ }
+
+ public Duration taskStartStopDelay() {
+ return randomNormalishJitter("100ms", "1s");
+ }
+ public Duration taskWorkflowDelay() {
+ return Duration.of("10s").add(randomNormalishJitter("0", "5s"));
+ }
+
+ @Test(groups="Integration") // because slow
+ public void testRace() throws Exception {
+ BasicApplication app = EntityManagementUtils.createUnstarted(mgmt, EntitySpec.create(BasicApplication.class));
+ int N=5;
+ // at N=5 we see: Recompute determined BasicStartableImpl{id=k7meoqgkqg} is up, after 17s 391ms
+ for (int i=0; i<N; i++) app.addChild(descendantSpec(N, N));
+
+// Dumper.dumpInfo(app);
+
+ decorateExecutionWithChaosMonkeySleepAndLog((BasicExecutionManager) mgmt.getExecutionManager(),
+ () -> onTaskStart(this::taskStartStopDelay), () -> onTaskEnd(this::taskStartStopDelay));
+
+ recurse(app, Entity::getChildren, this::initStrict);
+ app.start(null);
+
+ // stick a breakpoint on the following line (make sure it is thread-only, not all-threads!)
+ // then connect a UI eg brooklyn-ui/app-inspector `make dev` to the API endpoint used
+ decorateExecutionWithChaosMonkeySleepAndLog((BasicExecutionManager) mgmt.getExecutionManager(), null, null);
+ EntityAsserts.assertAttributeEquals(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.RUNNING);
+ }
+
+ private <T> void recurse(T root, Function<T,? extends Iterable<T>> next, Consumer<T> action) {
+ action.accept(root);
+ next.apply(root).forEach(child -> recurse(child, next, action));
+ }
+
+ private void initStrict(Entity target) {
+ if (target.getChildren().isEmpty()) return;
+
+ // some things set this, but not in this test
+// ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(target, Attributes.SERVICE_STATE_ACTUAL,
+// "created but not yet started, at "+Time.makeDateString());
+
+ ServiceStateLogic.newEnricherFromChildren().checkChildrenAndMembers()
+ .uniqueTag("children-service-up-contraindicators")
+ .requireUpChildren(QuorumCheck.QuorumChecks.allAndAtLeastOne())
+ .configure(ServiceStateLogic.ComputeServiceIndicatorsFromChildrenAndMembers.DERIVE_SERVICE_PROBLEMS, false)
+ .configure(ServiceStateLogic.ComputeServiceIndicatorsFromChildrenAndMembers.IGNORE_ENTITIES_WITH_THESE_SERVICE_STATES,
+ // override _not_ to ignore "starting" children
+ // (only applies for collections, which shouldn't say up unless starting children are up;
+ // other nodes default to an "alwaysHealthy" quorum so won't be affected,
+ // unless they have a custom up-ness quorum-check, in which case their upness will also be a function of starting children;
+ // means eg restarting a child of a collection will make the collection say service down, which is wanted for it;
+ // ideally this might be configurable for other nodes or depending on state of collection,
+ // or we can tie in with state derivation ... but that can be an enhancement for another day)
+ ImmutableSet.of(
+ //Lifecycle.STARTING, Lifecycle.STARTED,
+ //Lifecycle.STOPPING, Lifecycle.STOPPED,
+ Lifecycle.DESTROYED))
+ .addTo(target);
+
+ ServiceStateLogic.newEnricherFromChildren().checkChildrenAndMembers()
+ .uniqueTag("children-service-problems-indicators")
+ .requireRunningChildren(QuorumCheck.QuorumChecks.allAndAtLeastOne())
+ .configure(ServiceStateLogic.ComputeServiceIndicatorsFromChildrenAndMembers.DERIVE_SERVICE_NOT_UP, false)
+ .addTo(target);
+
+ target.enrichers().add(ServiceStateLogic.newEnricherForServiceStateFromProblemsAndUp());
+ target.enrichers().add(ServiceStateLogic.ServiceNotUpLogic.newEnricherForServiceUpIfNotUpIndicatorsEmpty());
+ }
+
+ private EntitySpec descendantSpec(int... sizes) {
+ if (sizes.length==0) return
+ EntitySpec.create(WorkflowStartable.class)
+ .configure((ConfigKey<Object>)(ConfigKey) WorkflowStartable.START_WORKFLOW, MutableMap.of(
+ "steps", MutableList.of("sleep "+taskWorkflowDelay().toMilliseconds()+"ms")))
+ .configure((ConfigKey<Object>)(ConfigKey) WorkflowStartable.STOP_WORKFLOW, MutableMap.of(
+ "steps", MutableList.of("let x = 0")));
+ EntitySpec<BasicStartable> result = EntitySpec.create(BasicStartable.class);
+ int[] nextSizes = new int[sizes.length-1];
+ for (int i=1; i<sizes.length; i++) nextSizes[i-1] = sizes[i];
+ for (int i=0; i<sizes[0]; i++) result.child(descendantSpec(nextSizes));
+ return result;
+ }
+
+ private void decorateExecutionWithChaosMonkeySleepAndLog(BasicExecutionManager ctx, Runnable onTaskStart, Runnable onTaskEnd) {
+ ctx.getAutoFlagsLive().put(BasicExecutionManager.TASK_START_CALLBACK_TAG, onTaskStart);
+ ctx.getAutoFlagsLive().put(BasicExecutionManager.TASK_END_CALLBACK_TAG, onTaskEnd);
+ }
+
+ private boolean includeForTaskDecoration(Task t) {
+ if (t==null) return false;
+ if (!Tasks.isNonProxyTask(t)) return false;
+ if (t.getDisplayName().equals("periodic-persister"))
+ return false;
+ return true;
+ }
+
+ private void onTaskStart(Supplier<Duration> delay) {
+ Task<?> current = BasicExecutionManager.getPerThreadCurrentTask().get();
+ if (includeForTaskDecoration(current)) {
+ // above necessary to prevent this running from
+ Duration d = delay.get();
+// log.info("TASK " + current+" delay "+d+" before start");
+ Time.sleep(d);
+// log.info("TASK " + current+" start");
+ }
+ }
+ private void onTaskEnd(Supplier<Duration> delay) {
+ Task<?> current = BasicExecutionManager.getPerThreadCurrentTask().get();
+ if (includeForTaskDecoration(current)) {
+ // above necessary to prevent this running from
+// log.info("TASK " + current+" delay before end");
+ Time.sleep(delay.get());
+// log.info("TASK " + current+" end");
+ }
+ }
+
+}