Adds AsyncApplication, and tests
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/13c81772 Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/13c81772 Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/13c81772 Branch: refs/heads/master Commit: 13c817720e82984f120d2356c7f9722560cd133b Parents: b724778 Author: Aled Sage <[email protected]> Authored: Thu Jul 6 16:07:15 2017 +0100 Committer: Aled Sage <[email protected]> Committed: Fri Jul 14 14:13:14 2017 +0100 ---------------------------------------------------------------------- .../core/entity/AbstractApplication.java | 2 +- .../brooklyn/entity/stock/AsyncApplication.java | 32 ++ .../entity/stock/AsyncApplicationImpl.java | 479 +++++++++++++++++++ .../brooklyn/core/entity/EntityAsyncTest.java | 361 ++++++++++++++ .../entity/stock/AsyncApplicationTest.java | 457 ++++++++++++++++++ 5 files changed, 1330 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/13c81772/core/src/main/java/org/apache/brooklyn/core/entity/AbstractApplication.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/entity/AbstractApplication.java b/core/src/main/java/org/apache/brooklyn/core/entity/AbstractApplication.java index 3cbee66..3a36c75 100644 --- a/core/src/main/java/org/apache/brooklyn/core/entity/AbstractApplication.java +++ b/core/src/main/java/org/apache/brooklyn/core/entity/AbstractApplication.java @@ -213,7 +213,7 @@ public abstract class AbstractApplication extends AbstractEntity implements Star } } - private static class ProblemStartingChildrenException extends RuntimeException { + protected static class ProblemStartingChildrenException extends RuntimeException { private static final long serialVersionUID = 7710856289284536803L; private ProblemStartingChildrenException(Exception cause) { super(cause); } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/13c81772/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplication.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplication.java b/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplication.java new file mode 100644 index 0000000..16ffaa4 --- /dev/null +++ b/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplication.java @@ -0,0 +1,32 @@ +/* + * 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.entity.stock; + +import org.apache.brooklyn.api.entity.ImplementedBy; +import org.apache.brooklyn.core.entity.StartableApplication; +import org.apache.brooklyn.core.entity.trait.AsyncStartable; + +/** + * An app that starts up asynchronously. Calling start will call start on its children, + * but it does not expect the children to have started by the time the start() effector + * has returned. Instead, it infers from the children's state whether they are up or not. + */ +@ImplementedBy(AsyncApplicationImpl.class) +public interface AsyncApplication extends StartableApplication, AsyncStartable { +} http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/13c81772/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplicationImpl.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplicationImpl.java b/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplicationImpl.java new file mode 100644 index 0000000..636ab4f --- /dev/null +++ b/core/src/main/java/org/apache/brooklyn/entity/stock/AsyncApplicationImpl.java @@ -0,0 +1,479 @@ +/* + * 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.entity.stock; + +import static org.apache.brooklyn.core.entity.Attributes.SERVICE_STATE_ACTUAL; +import static org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.clearMapSensorEntry; +import static org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.updateMapSensorEntry; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.brooklyn.api.entity.Entity; +import org.apache.brooklyn.api.entity.EntityLocal; +import org.apache.brooklyn.api.location.Location; +import org.apache.brooklyn.api.sensor.EnricherSpec; +import org.apache.brooklyn.api.sensor.Sensor; +import org.apache.brooklyn.api.sensor.SensorEvent; +import org.apache.brooklyn.api.sensor.SensorEventListener; +import org.apache.brooklyn.config.ConfigKey; +import org.apache.brooklyn.core.BrooklynLogging; +import org.apache.brooklyn.core.config.BasicConfigInheritance; +import org.apache.brooklyn.core.config.ConfigKeys; +import org.apache.brooklyn.core.entity.AbstractApplication; +import org.apache.brooklyn.core.entity.Attributes; +import org.apache.brooklyn.core.entity.Entities; +import org.apache.brooklyn.core.entity.lifecycle.Lifecycle; +import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic; +import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceProblemsLogic; +import org.apache.brooklyn.enricher.stock.AbstractMultipleSensorAggregator; +import org.apache.brooklyn.util.collections.MutableList; +import org.apache.brooklyn.util.collections.MutableSet; +import org.apache.brooklyn.util.collections.QuorumCheck; +import org.apache.brooklyn.util.exceptions.Exceptions; +import org.apache.brooklyn.util.text.Strings; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.MoreObjects; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.common.reflect.TypeToken; + +public class AsyncApplicationImpl extends AbstractApplication implements AsyncApplication { + + private static final Logger LOG = LoggerFactory.getLogger(AsyncApplicationImpl.class); + + @Override + public void init() { + // Code below copied from BasicAppliationImpl. + // Set the default name *before* calling super.init(), and only do so if we don't have an + // explicit default. This is a belt-and-braces fix: before we overwrote the defaultDisplayName + // that was inferred from the catalog item name. + if (Strings.isBlank(getConfig(DEFAULT_DISPLAY_NAME))) { + setDefaultDisplayName("Application ("+getId()+")"); + } + super.init(); + } + + @Override + protected void initEnrichers() { + // Deliberately not calling `super.initEnrichers()`. For our state (i.e. "service.state" + // and "service.isUp"), we rely on the `serviceStateComputer`. This keeps things a lot + // simpler. However, it means that if someone manually sets a "service.notUp.indicators" + // or "service.problems" then that won't cause the entity to transition to false or ON_FIRE. + + enrichers().add(EnricherSpec.create(ServiceStateComputer.class) + .configure(ServiceStateComputer.FROM_CHILDREN, true) + .configure(ServiceStateComputer.UP_QUORUM_CHECK, config().get(UP_QUORUM_CHECK)) + .configure(ServiceStateComputer.RUNNING_QUORUM_CHECK, config().get(RUNNING_QUORUM_CHECK))); + + } + + // Only overrides AbstractApplication.start so as to disable the publishing of expected=running. + // Code is copy-pasted verbatim from AbstractAppliation, except for not setting Lifecycle.RUNNING. + @Override + public void start(Collection<? extends Location> locations) { + this.addLocations(locations); + // 2016-01: only pass locations passed to us, as per ML discussion + Collection<? extends Location> locationsToUse = locations==null ? ImmutableSet.<Location>of() : locations; + ServiceProblemsLogic.clearProblemsIndicator(this, START); + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, Attributes.SERVICE_STATE_ACTUAL, "Application starting"); + ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(this, START.getName()); + setExpectedStateAndRecordLifecycleEvent(Lifecycle.STARTING); + try { + try { + + preStart(locationsToUse); + + // Opportunity to block startup until other dependent components are available + Object val = config().get(START_LATCH); + if (val != null) LOG.debug("{} finished waiting for start-latch; continuing...", this); + + doStart(locationsToUse); + postStart(locationsToUse); + + } catch (ProblemStartingChildrenException e) { + throw Exceptions.propagate(e); + } catch (Exception e) { + // should remember problems, apart from those that happened starting children + // fixed bug introduced by the fix in dacf18b831e1e5e1383d662a873643a3c3cabac6 + // where failures in special code at application root don't cause app to go on fire + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), Exceptions.collapseText(e)); + throw Exceptions.propagate(e); + } + + } catch (Exception e) { + recordApplicationEvent(Lifecycle.ON_FIRE); + ServiceStateLogic.setExpectedStateRunningWithErrors(this); + + // no need to log here; the effector invocation should do that + throw Exceptions.propagate(e); + + } finally { + ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(this, Attributes.SERVICE_STATE_ACTUAL); + } + + // CHANGE FROM SUPER: NOT CALLING THESE + // ServiceStateLogic.setExpectedState(this, Lifecycle.RUNNING); + // setExpectedStateAndRecordLifecycleEvent(Lifecycle.RUNNING); + // + // logApplicationLifecycle("Started"); + } + + /** + * Calculates the "service.state" and "service.isUp", based on the state of the children. It + * also transitions from expected=starting to expected=running once all of the children have + * finished their starting. + * + * This does <em>not</em> just rely on the "service.problems" and "service.notUp.indicators" + * because those make different assumptions about the expected state. Instead it seems much + * easier to implement the specific logic for async startup here. + * + * The important part of the implementation is {@link #onUpdated()}, and its helper methods + * for {@link #computeServiceUp(Lifecycle)}, {@link #computeServiceState(Lifecycle)} and + * {@link #computeExpectedState(Lifecycle, Lifecycle)}. + * + * This class is not to be instantiated directly. Instead, if cusotmization is desired then override + * {@link AsyncApplicationImpl#initEnrichers()} to create and add this enricher (with the same unique + * tag, to replace the default). + */ + public static class ServiceStateComputer extends AbstractMultipleSensorAggregator<Void> implements SensorEventListener<Object> { + /** standard unique tag identifying instances of this enricher at runtime */ + public final static String DEFAULT_UNIQUE_TAG = "async-service-state-computer"; + + public static final ConfigKey<QuorumCheck> RUNNING_QUORUM_CHECK = ConfigKeys.builder(QuorumCheck.class, "runningQuorumCheck") + .description("Logic for checking whether this service is running, based on children and/or " + + "members running (by default requires all, but ignores any that are stopping)") + .defaultValue(QuorumCheck.QuorumChecks.all()) + .runtimeInheritance(BasicConfigInheritance.NOT_REINHERITED) + .build(); + + public static final ConfigKey<QuorumCheck> UP_QUORUM_CHECK = ConfigKeys.builder(QuorumCheck.class, "upQuorumCheck") + .description("Logic for checking whether this service is up, based on children and/or members (by default requires all)") + .defaultValue(QuorumCheck.QuorumChecks.all()) + .runtimeInheritance(BasicConfigInheritance.NOT_REINHERITED) + .build(); + + // TODO How does this relate to quorum?! + @SuppressWarnings("serial") + public static final ConfigKey<Set<Lifecycle>> ENTITY_FAILED_STATES = ConfigKeys.builder(new TypeToken<Set<Lifecycle>>() {}) + .name("entityFailedStates") + .description("Service states that indicate a child/member has failed (by default just ON_FIRE will mean not healthy)") + .defaultValue(ImmutableSet.of(Lifecycle.ON_FIRE)) + .runtimeInheritance(BasicConfigInheritance.NOT_REINHERITED) + .build(); + + @SuppressWarnings("serial") + public static final ConfigKey<Set<Lifecycle>> ENTITY_TRANSITION_STATES_ON_STARTING = ConfigKeys.builder(new TypeToken<Set<Lifecycle>>() {}) + .name("entityTransitionStatesOnStarting") + .description("Service states which indicate a child/member is still starting " + + "(used to compute when we have finished starting)") + .defaultValue(MutableSet.of(null, Lifecycle.CREATED, Lifecycle.STARTING).asUnmodifiable()) + .build(); + + @SuppressWarnings("serial") + public static final ConfigKey<Set<Lifecycle>> ENTITY_IGNORED_STATES_ON_STARTING = ConfigKeys.builder(new TypeToken<Set<Lifecycle>>() {}) + .name("entityIgnoredStatesOnStarting") + .description("Service states of a child/member that mean we'll ignore it, for calculating " + + "our own state when 'staring' (by default ignores children that are stopping/stopped)") + .defaultValue(ImmutableSet.of(Lifecycle.STOPPING, Lifecycle.STOPPED, Lifecycle.DESTROYED)) + .build(); + + @SuppressWarnings("serial") + public static final ConfigKey<Set<Lifecycle>> ENTITY_IGNORED_STATES_ON_OTHERS = ConfigKeys.builder(new TypeToken<Set<Lifecycle>>() {}) + .name("entityIgnoredStatesOnOthers") + .description("Service states of a child/member that mean we'll ignore it, for calculating " + + "our own state when we are not 'staring' (by default ignores children that are starting/stopping)") + .defaultValue(MutableSet.of(null, Lifecycle.STOPPING, Lifecycle.STOPPED, Lifecycle.DESTROYED, Lifecycle.CREATED, Lifecycle.STARTING).asUnmodifiable()) + .build(); + + public static final ConfigKey<Boolean> IGNORE_ENTITIES_WITH_SERVICE_UP_NULL = ConfigKeys.builder(Boolean.class) + .name("ignoreEntitiesWithServiceUpNull") + .description("Whether to ignore children reporting null values for service up " + + "(i.e. don't treat them as 'down' when computing our own 'service.isUp')") + .defaultValue(true) + .build(); + + static final Set<ConfigKey<?>> RECONFIGURABLE_KEYS = ImmutableSet.<ConfigKey<?>>of( + UP_QUORUM_CHECK, RUNNING_QUORUM_CHECK, + ENTITY_IGNORED_STATES_ON_STARTING, ENTITY_IGNORED_STATES_ON_OTHERS, + ENTITY_FAILED_STATES, ENTITY_TRANSITION_STATES_ON_STARTING); + + static final List<Sensor<?>> SOURCE_SENSORS = ImmutableList.<Sensor<?>>of(SERVICE_UP, SERVICE_STATE_ACTUAL); + + @Override + public AsyncApplicationImpl getEntity() { + return (AsyncApplicationImpl) super.getEntity(); + } + + @Override + protected void setEntityLoadingTargetConfig() { + // ensure parent's behaviour never happens + if (getConfig(TARGET_SENSOR)!=null) + throw new IllegalArgumentException("Must not set "+TARGET_SENSOR+" when using "+this); + } + + @Override + public void setEntity(EntityLocal entity) { + if (!(entity instanceof AsyncApplicationImpl)) { + throw new IllegalArgumentException("enricher designed to work only with async-apps"); + } + if (!isRebinding() && Boolean.FALSE.equals(config().get(SUPPRESS_DUPLICATES))) { + throw new IllegalArgumentException("Must not set "+SUPPRESS_DUPLICATES+" to false when using "+this); + } + super.setEntity(entity); + if (suppressDuplicates==null) { + // only publish on changes, unless it is configured otherwise + suppressDuplicates = true; + } + + // Need to update again, e.g. if stop() effector marks this as expected=stopped. + // There'd be a risk of infinite loop if we didn't suppressDuplicates! + subscriptions().subscribe(entity, Attributes.SERVICE_STATE_EXPECTED, new SensorEventListener<Lifecycle.Transition>() { + @Override public void onEvent(SensorEvent<Lifecycle.Transition> event) { + onUpdated(); + }}); + } + + @Override + protected <T> void doReconfigureConfig(ConfigKey<T> key, T val) { + if (RECONFIGURABLE_KEYS.contains(key)) { + return; + } else { + super.doReconfigureConfig(key, val); + } + } + + @Override + protected void onChanged() { + super.onChanged(); + onUpdated(); + } + + @Override + protected Collection<Sensor<?>> getSourceSensors() { + return SOURCE_SENSORS; + } + + @Override + protected void onUpdated() { + if (entity == null || !isRunning() || !Entities.isManaged(entity)) { + // e.g. invoked during setup or entity has become unmanaged; just ignore + BrooklynLogging.log(LOG, BrooklynLogging.levelDebugOrTraceIfReadOnly(entity), + "Ignoring {} onUpdated when entity is not in valid state ({})", this, entity); + return; + } + + Lifecycle.Transition oldExpectedStateTransition = entity.sensors().get(Attributes.SERVICE_STATE_EXPECTED); + Lifecycle oldExpectedState = (oldExpectedStateTransition != null) ? oldExpectedStateTransition.getState() : null; + + ValueAndReason<Boolean> newServiceUp = computeServiceUp(oldExpectedState); + ValueAndReason<Lifecycle> newServiceState = computeServiceState(oldExpectedState); + Lifecycle newExpectedState = computeExpectedState(oldExpectedState, newServiceState.val); + + emit(Attributes.SERVICE_STATE_ACTUAL, newServiceState.val); + emit(Attributes.SERVICE_UP, newServiceUp.val); + + if (Boolean.TRUE.equals(newServiceUp.val)) { + clearMapSensorEntry(entity, Attributes.SERVICE_NOT_UP_INDICATORS, DEFAULT_UNIQUE_TAG); + } else { + updateMapSensorEntry(entity, Attributes.SERVICE_NOT_UP_INDICATORS, DEFAULT_UNIQUE_TAG, newServiceUp.reason); + } + if (newServiceState.val != null && newServiceState.val == Lifecycle.ON_FIRE) { + updateMapSensorEntry(entity, Attributes.SERVICE_PROBLEMS, DEFAULT_UNIQUE_TAG, newServiceState.reason); + } else { + clearMapSensorEntry(entity, Attributes.SERVICE_PROBLEMS, DEFAULT_UNIQUE_TAG); + } + + if (oldExpectedState != newExpectedState) { + // TODO could check no-one else has changed expectedState (e.g. by calling "stop") + // TODO do we need to subscribe to our own serviceStateExpected, in case someone calls stop? + getEntity().setExpectedStateAndRecordLifecycleEvent(newExpectedState); + } + } + + /** + * Count the entities that are up versus not up. Compare this with the quorum required. + */ + protected ValueAndReason<Boolean> computeServiceUp(Lifecycle expectedState) { + boolean ignoreNull = getConfig(IGNORE_ENTITIES_WITH_SERVICE_UP_NULL); + Set<Lifecycle> ignoredStates; + if (expectedState == Lifecycle.STARTING) { + ignoredStates = getConfig(ENTITY_IGNORED_STATES_ON_STARTING); + } else { + ignoredStates = getConfig(ENTITY_IGNORED_STATES_ON_OTHERS); + } + + Map<Entity, Boolean> values = getValues(SERVICE_UP); + List<Entity> violators = MutableList.of(); + + int entries = 0; + int numUp = 0; + for (Map.Entry<Entity, Boolean> entry : values.entrySet()) { + Lifecycle entityState = entry.getKey().getAttribute(SERVICE_STATE_ACTUAL); + + if (ignoreNull && entry.getValue()==null) { + continue; + } + if (ignoredStates.contains(entityState)) { + continue; + } + entries++; + + if (Boolean.TRUE.equals(entry.getValue())) { + numUp++; + } else { + violators.add(entry.getKey()); + } + } + + QuorumCheck qc = getRequiredConfig(UP_QUORUM_CHECK); + if (qc.isQuorate(numUp, violators.size()+numUp)) { + // quorate + return new ValueAndReason<>(Boolean.TRUE, "quorate"); + } + + String reason; + if (values.isEmpty()) { + reason = "No entities present"; + } else if (entries == 0) { + reason = "No entities (in correct state) publishing service up"; + } else if (violators.isEmpty()) { + reason = "Not enough entities"; + } else if (violators.size() == 1) { + reason = violators.get(0)+" is not up"; + } else if (violators.size() == entries) { + reason = "None of the entities are up"; + } else { + reason = violators.size()+" entities are not up, including "+violators.get(0); + } + return new ValueAndReason<>(Boolean.FALSE, reason); + } + + protected ValueAndReason<Lifecycle> computeServiceState(Lifecycle expectedState) { + if (expectedState != null && (expectedState != Lifecycle.STARTING && expectedState != Lifecycle.RUNNING)) { + return new ValueAndReason<>(expectedState, "expected state "+expectedState); + } + + Set<Lifecycle> ignoredStates; + if (expectedState == Lifecycle.STARTING) { + ignoredStates = getConfig(ENTITY_IGNORED_STATES_ON_STARTING); + } else { + ignoredStates = getConfig(ENTITY_IGNORED_STATES_ON_OTHERS); + } + Set<Lifecycle> transitionStates; + if (expectedState == Lifecycle.STARTING) { + transitionStates = getConfig(ENTITY_TRANSITION_STATES_ON_STARTING); + } else { + transitionStates = ImmutableSet.of(); + } + + Map<Entity, Lifecycle> values = getValues(SERVICE_STATE_ACTUAL); + List<Entity> violators = MutableList.of(); + int entries = 0; + int numRunning = 0; + int numTransitioning = 0; + + for (Map.Entry<Entity,Lifecycle> entry : values.entrySet()) { + if (ignoredStates.contains(entry.getValue())) { + continue; + } + entries++; + + if (entry.getValue() == Lifecycle.RUNNING) { + numRunning++; + } else if (transitionStates.contains(entry.getValue())) { + numTransitioning++; + } else { + violators.add(entry.getKey()); + } + } + + QuorumCheck qc = getConfig(RUNNING_QUORUM_CHECK); + if (qc.isQuorate(numRunning, violators.size()+numRunning+numTransitioning)) { + // quorate + return new ValueAndReason<Lifecycle>(Lifecycle.RUNNING, "quorate"); + } + boolean canEverBeQuorate = qc.isQuorate(numRunning+numTransitioning, violators.size()+numRunning+numTransitioning); + if (expectedState == Lifecycle.STARTING && canEverBeQuorate) { + // quorate + return new ValueAndReason<Lifecycle>(Lifecycle.STARTING, "not yet quorate"); + } + + String reason; + if (values.isEmpty()) { + reason = "No entities present"; + } else if (entries == 0) { + reason = "No entities in interesting states"; + } else if (violators.isEmpty()) { + reason = "Not enough entities"; + } else if (violators.size() == 1) { + reason = violators.get(0)+" is not healthy"; + } else if (violators.size() == entries) { + reason = "None of the entities are healthy"; + } else { + reason = violators.size()+" entities are not healthy, including "+violators.get(0); + } + return new ValueAndReason<>(Lifecycle.ON_FIRE, reason); + + } + + protected Lifecycle computeExpectedState(Lifecycle oldExpectedState, Lifecycle newActualState) { + if (oldExpectedState != Lifecycle.STARTING) { + return oldExpectedState; + } + + // Are any of our children still starting? + Map<Entity, Lifecycle> values = getValues(SERVICE_STATE_ACTUAL); + boolean childIsStarting = values.containsValue(Lifecycle.STARTING) || values.containsValue(Lifecycle.CREATED) || values.containsValue(null); + + if (!childIsStarting) { + // We only transition to expected=RUNNING if all our children are no longer starting. + // When actual=running, this matters if quorum != all; + // When actual=on_fire, this matters for recovering. + return Lifecycle.RUNNING; + } + return oldExpectedState; + } + + /** not used; see specific `computeXxx` methods, invoked by overridden onUpdated */ + @Override + protected Object compute() { + return null; + } + + static class ValueAndReason<T> { + final T val; + final String reason; + + ValueAndReason(T val, String reason) { + this.val = val; + this.reason = reason; + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("val", val).add("reason", reason).toString(); + } + } + } +} http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/13c81772/core/src/test/java/org/apache/brooklyn/core/entity/EntityAsyncTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/entity/EntityAsyncTest.java b/core/src/test/java/org/apache/brooklyn/core/entity/EntityAsyncTest.java new file mode 100644 index 0000000..0a59867 --- /dev/null +++ b/core/src/test/java/org/apache/brooklyn/core/entity/EntityAsyncTest.java @@ -0,0 +1,361 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.brooklyn.core.entity; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +import java.util.Collection; +import java.util.List; + +import org.apache.brooklyn.api.entity.Entity; +import org.apache.brooklyn.api.entity.EntitySpec; +import org.apache.brooklyn.api.entity.ImplementedBy; +import org.apache.brooklyn.api.location.Location; +import org.apache.brooklyn.core.entity.lifecycle.Lifecycle; +import org.apache.brooklyn.core.entity.lifecycle.Lifecycle.Transition; +import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic; +import org.apache.brooklyn.core.entity.trait.Startable; +import org.apache.brooklyn.core.entity.trait.StartableMethods; +import org.apache.brooklyn.core.test.BrooklynAppUnitTestSupport; +import org.apache.brooklyn.entity.group.AbstractGroupImpl; +import org.apache.brooklyn.entity.group.Cluster; +import org.apache.brooklyn.test.Asserts; +import org.apache.brooklyn.util.collections.QuorumCheck; +import org.apache.brooklyn.util.exceptions.Exceptions; +import org.apache.brooklyn.util.time.Duration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.Test; + +import com.google.common.base.Optional; +import com.google.common.base.Predicate; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; + +/** + * Tests a pattern that a user has, for async entities. + * + * Calling start() triggers some asynchronous work. Completion of that work is reported via a + * callback that indicates success or fail. + */ +public class EntityAsyncTest extends BrooklynAppUnitTestSupport { + + // TODO If the cluster has quorum=all, should the cluster report ON_FIRE as soon as any of the + // children report a failure to start, or should it keep saying "STARTING" until all callbacks + // are received? + + private static final Logger LOG = LoggerFactory.getLogger(EntityAsyncTest.class); + + @Test + public void testEntityStartsAsynchronously() throws Exception { + AsyncEntity entity = app.addChild(EntitySpec.create(AsyncEntity.class)); + + app.start(ImmutableList.of()); + assertStateEventually(entity, Lifecycle.STARTING, Lifecycle.STARTING, false); + + entity.onCallback(true); + assertStateEventually(entity, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testClusterStartsAsynchronously() throws Exception { + AsyncCluster cluster = app.addChild(EntitySpec.create(AsyncCluster.class) + .configure(AsyncCluster.INITIAL_SIZE, 2)); + + app.start(ImmutableList.of()); + List<AsyncEntity> children = cast(cluster.getChildren(), AsyncEntity.class); + + // Everything should say "starting" + assertStateEventually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + for (AsyncEntity child : children) { + assertStateEventually(child, Lifecycle.STARTING, Lifecycle.STARTING, false); + } + + // Indicate that first child is now running successfully + cluster.onCallback(children.get(0).getId(), true); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateContinually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Indicate that second child is now running successfully + cluster.onCallback(children.get(1).getId(), true); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testClusterFirstChildFails() throws Exception { + AsyncCluster cluster = app.addChild(EntitySpec.create(AsyncCluster.class) + .configure(AsyncCluster.INITIAL_SIZE, 2)); + + app.start(ImmutableList.of()); + List<AsyncEntity> children = cast(cluster.getChildren(), AsyncEntity.class); + + // Everything should say "starting" + assertStateEventually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + for (AsyncEntity child : children) { + assertStateEventually(child, Lifecycle.STARTING, Lifecycle.STARTING, false); + } + + // Indicate that first child failed + cluster.onCallback(children.get(0).getId(), false); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(children.get(1), Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Indicate that second child is now running successfully + cluster.onCallback(children.get(1).getId(), true); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testClusterSecondChildFails() throws Exception { + AsyncCluster cluster = app.addChild(EntitySpec.create(AsyncCluster.class) + .configure(AsyncCluster.INITIAL_SIZE, 2)); + + app.start(ImmutableList.of()); + List<AsyncEntity> children = cast(cluster.getChildren(), AsyncEntity.class); + + // Everything should say "starting" + assertStateEventually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + for (AsyncEntity child : children) { + assertStateEventually(child, Lifecycle.STARTING, Lifecycle.STARTING, false); + } + + // Indicate that first child is now running successfully + cluster.onCallback(children.get(0).getId(), true); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateContinually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Indicate that second child failed + cluster.onCallback(children.get(1).getId(), false); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testClusterStartsThenChildFails() throws Exception { + AsyncCluster cluster = app.addChild(EntitySpec.create(AsyncCluster.class) + .configure(AsyncCluster.INITIAL_SIZE, 2)); + + app.start(ImmutableList.of()); + List<AsyncEntity> children = cast(cluster.getChildren(), AsyncEntity.class); + + // Everything should say "starting" + assertStateEventually(cluster, Lifecycle.STARTING, Lifecycle.STARTING, false); + for (AsyncEntity child : children) { + assertStateEventually(child, Lifecycle.STARTING, Lifecycle.STARTING, false); + } + + // Indicate that children are now running successfully + cluster.onCallback(children.get(0).getId(), true); + cluster.onCallback(children.get(1).getId(), true); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + + // First child then fails + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(children.get(0), "myKey", "simulate failure"); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + + // First child then recovers + ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(children.get(0), "myKey"); + + assertStateEventually(children.get(0), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(children.get(1), Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(cluster, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + private void assertStateContinually(final Entity entity, final Lifecycle expectedState, final Lifecycle state, final Boolean isUp) { + Asserts.succeedsContinually(ImmutableMap.of("timeout", Duration.millis(50)), new Runnable() { + @Override public void run() { + assertState(entity, expectedState, state, isUp); + }}); + } + + private void assertStateEventually(Entity entity, Lifecycle expectedState, Lifecycle state, Boolean isUp) { + Asserts.succeedsEventually(new Runnable() { + @Override public void run() { + assertState(entity, expectedState, state, isUp); + }}); + } + + private void assertState(Entity entity, Lifecycle expectedState, Lifecycle state, Boolean isUp) { + Transition actualExpectedState = entity.sensors().get(Attributes.SERVICE_STATE_EXPECTED); + Lifecycle actualState = entity.sensors().get(Attributes.SERVICE_STATE_ACTUAL); + Boolean actualIsUp = entity.sensors().get(Attributes.SERVICE_UP); + String msg = "actualExpectedState="+actualExpectedState+", actualState="+actualState+", actualIsUp="+actualIsUp; + if (expectedState != null) { + assertEquals(actualExpectedState.getState(), expectedState, msg); + } else { + assertTrue(actualExpectedState == null || actualExpectedState.getState() == null, msg); + } + assertEquals(actualState, state, msg); + assertEquals(actualIsUp, isUp, msg); + } + + private <T> List<T> cast(Iterable<?> vals, Class<T> clazz) { + List<T> result = Lists.newArrayList(); + for (Object val : vals) { + result.add(clazz.cast(val)); + } + return result; + } + + /** + * The AsyncEntity's start leaves it in a "STARTING" state. + * + * It stays like that until {@code onCallback(true)} is called. It should then report + * expected="RUNNING", service.state="RUNNING" and service.isUp=true. Alternatively, + * if {@code onCallback(true)} is called, it should then report expected="RUNNING", + * service.state="ON_FIRE" and service.isUp=false. + */ + @ImplementedBy(AsyncEntityImpl.class) + public interface AsyncEntity extends Entity, Startable { + void onCallback(boolean success); + } + + public static class AsyncEntityImpl extends AbstractEntity implements AsyncEntity { + + @Override + public void start(Collection<? extends Location> locations) { + ServiceStateLogic.setExpectedState(this, Lifecycle.STARTING); + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), "starting"); + } + + + @Override + public void onCallback(boolean success) { + if (success) { + ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(this, START.getName()); + } else { + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), "callback reported failure"); + } + + Transition expectedState = sensors().get(Attributes.SERVICE_STATE_EXPECTED); + if (expectedState != null && expectedState.getState() == Lifecycle.STARTING) { + ServiceStateLogic.setExpectedState(this, Lifecycle.RUNNING); + } + } + + @Override + public void stop() { + } + + @Override + public void restart() { + } + } + + /** + * The AsyncCluster's start leaves it in a "STARTING" state, having created the children + * and called start() on each. + * + * It expects an explicit call to {@code clearNotUpIndicator(); setExpected(Lifecycle.RUNNING)}, + * after which its service.isUp will be inferred from its children (they must all be running). + */ + @ImplementedBy(AsyncClusterImpl.class) + public interface AsyncCluster extends Cluster, Startable { + void onCallback(String childId, boolean success); + } + + public static class AsyncClusterImpl extends AbstractGroupImpl implements AsyncCluster { + + @Override + protected void initEnrichers() { + super.initEnrichers(); + + // all children must be up, for ourselves to be up + enrichers().add(ServiceStateLogic.newEnricherFromChildrenUp() + .checkChildrenOnly() + .requireUpChildren(QuorumCheck.QuorumChecks.all())); + } + + @Override + public void start(Collection<? extends Location> locations) { + ServiceStateLogic.setExpectedState(this, Lifecycle.STARTING); + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), "starting"); + + try { + for (int i = 0; i < config().get(INITIAL_SIZE); i++) { + addChild(EntitySpec.create(AsyncEntity.class)); + } + StartableMethods.start(this, locations); + + } catch (Throwable t) { + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), Exceptions.collapseText(t)); + ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE); + throw Exceptions.propagate(t); + } + } + + @Override + public void onCallback(String childId, boolean success) { + Optional<Entity> child = Iterables.tryFind(getChildren(), EntityPredicates.idEqualTo(childId)); + if (child.isPresent()) { + ((AsyncEntity)child.get()).onCallback(success); + } else { + LOG.warn("Child not found with resourceId '"+childId+"'; not injecting state from callback"); + } + + Optional<Entity> unstartedVm = Iterables.tryFind(getChildren(), EntityPredicates.attributeSatisfies(Attributes.SERVICE_STATE_EXPECTED, + new Predicate<Lifecycle.Transition>() { + @Override public boolean apply(Transition input) { + return input == null || input.getState() == Lifecycle.STARTING; + }})); + + if (!unstartedVm.isPresent()) { + // No VMs are still starting; we are finished starting + ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(this, START.getName()); + ServiceStateLogic.setExpectedState(this, Lifecycle.RUNNING); + } + } + + @Override + public void stop() { + } + + @Override + public void restart() { + } + + @Override + public Integer resize(Integer desiredSize) { + throw new UnsupportedOperationException(); + } + } +} http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/13c81772/core/src/test/java/org/apache/brooklyn/entity/stock/AsyncApplicationTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/entity/stock/AsyncApplicationTest.java b/core/src/test/java/org/apache/brooklyn/entity/stock/AsyncApplicationTest.java new file mode 100644 index 0000000..8ca07cf --- /dev/null +++ b/core/src/test/java/org/apache/brooklyn/entity/stock/AsyncApplicationTest.java @@ -0,0 +1,457 @@ +/* + * 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.entity.stock; + +import static com.google.common.base.Preconditions.checkNotNull; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + +import java.util.Collection; + +import org.apache.brooklyn.api.entity.Entity; +import org.apache.brooklyn.api.entity.EntitySpec; +import org.apache.brooklyn.api.entity.ImplementedBy; +import org.apache.brooklyn.api.location.Location; +import org.apache.brooklyn.core.entity.AbstractEntity; +import org.apache.brooklyn.core.entity.Attributes; +import org.apache.brooklyn.core.entity.Entities; +import org.apache.brooklyn.core.entity.lifecycle.Lifecycle; +import org.apache.brooklyn.core.entity.lifecycle.Lifecycle.Transition; +import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic; +import org.apache.brooklyn.core.entity.trait.FailingEntity; +import org.apache.brooklyn.core.entity.trait.Startable; +import org.apache.brooklyn.core.test.BrooklynMgmtUnitTestSupport; +import org.apache.brooklyn.core.test.entity.TestEntity; +import org.apache.brooklyn.test.Asserts; +import org.apache.brooklyn.util.collections.QuorumCheck.QuorumChecks; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + +public class AsyncApplicationTest extends BrooklynMgmtUnitTestSupport { + + AsyncApplication app; + + @BeforeMethod(alwaysRun=true) + @Override + public void setUp() throws Exception { + super.setUp(); + app = mgmt.getEntityManager().createEntity(EntitySpec.create(AsyncApplication.class)); + } + + // FIXME This fails because the enricher is never triggered (there are no child events to trigger it) + @Test(enabled=false, groups="WIP") + public void testStartEmptyApp() throws Exception { + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartAndStopWithVanillaChild() throws Exception { + app = mgmt.getEntityManager().createEntity(EntitySpec.create(AsyncApplication.class) + .configure(AsyncApplication.DESTROY_ON_STOP, false)); + TestEntity child = app.addChild(EntitySpec.create(TestEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(child, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + + app.stop(); + + assertTrue(Entities.isManaged(app)); + assertTrue(Entities.isManaged(child)); + assertStateEventually(child, Lifecycle.STOPPED, Lifecycle.STOPPED, false); + assertStateEventually(app, Lifecycle.STOPPED, Lifecycle.STOPPED); + } + + @Test + public void testStopWillUnmanage() throws Exception { + TestEntity child = app.addChild(EntitySpec.create(TestEntity.class)); + app.start(ImmutableList.of()); + app.stop(); + + assertFalse(Entities.isManaged(app)); + assertFalse(Entities.isManaged(child)); + } + + @Test + public void testStartWithVanillaFailingChild() throws Exception { + app.addChild(EntitySpec.create(FailingEntity.class) + .configure(FailingEntity.FAIL_ON_START, true)); + try { + app.start(ImmutableList.of()); + Asserts.shouldHaveFailedPreviously(); + } catch (Exception e) { + Asserts.expectedFailureContains(e, "Simulating entity start failure for test"); + } + + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testStartWithAsyncChild() throws Exception { + AsyncEntity child = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + child.clearNotUpIndicators(); + child.setExpected(Lifecycle.RUNNING); + assertStateEventually(child, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithAsyncFailingChild() throws Exception { + AsyncEntity child = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + child.addNotUpIndicator("simulatedFailure", "my failure"); + child.setExpected(Lifecycle.RUNNING); + assertStateEventually(child, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testStartWithAsyncChildren() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child starts + child1.clearNotUpIndicators(); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Second child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithAsyncChildrenFirstChildFails() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child fails + child1.addNotUpIndicator("simulatedFailure", "my failure"); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.ON_FIRE, false); + + // Second child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testStartWithAsyncChildrenFirstChildFailsThenRecovers() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child fails + child1.addNotUpIndicator("simulatedFailure", "my failure"); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.ON_FIRE, false); + + // First child recovers + child1.clearNotUpIndicators(); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Second child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithAsyncChildrenFirstChildFailsThenAfterSecondItRecovers() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child fails + child1.addNotUpIndicator("simulatedFailure", "my failure"); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.ON_FIRE, false); + + // Second child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + + // First child recovers + child1.clearNotUpIndicators(); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithAsyncChildrenFirstChildFailsThenRecoversImmediately() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child fails + child1.addNotUpIndicator("simulatedFailure", "my failure"); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.ON_FIRE, false); + + // First child recovers + child1.clearNotUpIndicators(); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Second child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithAsyncChildrenLastChildFails() throws Exception { + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child starts + child1.clearNotUpIndicators(); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Seconds child fails + child2.addNotUpIndicator("simulatedFailure", "my failure"); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + } + + @Test + public void testStartWithQuorumOne() throws Exception { + app = mgmt.getEntityManager().createEntity(EntitySpec.create(AsyncApplication.class) + .configure(AsyncApplication.RUNNING_QUORUM_CHECK, QuorumChecks.atLeastOne()) + .configure(AsyncApplication.UP_QUORUM_CHECK, QuorumChecks.atLeastOne())); + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child starts + child1.clearNotUpIndicators(); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.RUNNING, true); + + // Seconds child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithQuorumOneFirstChildFails() throws Exception { + app = mgmt.getEntityManager().createEntity(EntitySpec.create(AsyncApplication.class) + .configure(AsyncApplication.RUNNING_QUORUM_CHECK, QuorumChecks.atLeastOne()) + .configure(AsyncApplication.UP_QUORUM_CHECK, QuorumChecks.atLeastOne())); + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child starts + child1.addNotUpIndicator("simulatedFailure", "my failure"); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // Seconds child starts + child2.clearNotUpIndicators(); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + @Test + public void testStartWithQuorumOneSecondChildFails() throws Exception { + app = mgmt.getEntityManager().createEntity(EntitySpec.create(AsyncApplication.class) + .configure(AsyncApplication.RUNNING_QUORUM_CHECK, QuorumChecks.atLeastOne()) + .configure(AsyncApplication.UP_QUORUM_CHECK, QuorumChecks.atLeastOne())); + AsyncEntity child1 = app.addChild(EntitySpec.create(AsyncEntity.class)); + AsyncEntity child2 = app.addChild(EntitySpec.create(AsyncEntity.class)); + app.start(ImmutableList.of()); + + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.STARTING, false); + + // First child starts + child1.clearNotUpIndicators(); + child1.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.STARTING, Lifecycle.STARTING, false); + assertStateEventually(app, Lifecycle.STARTING, Lifecycle.RUNNING, true); + + // Seconds child starts + child2.addNotUpIndicator("simulatedFailure", "my failure"); + child2.setExpected(Lifecycle.RUNNING); + assertStateEventually(child1, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + assertStateEventually(child2, Lifecycle.RUNNING, Lifecycle.ON_FIRE, false); + assertStateEventually(app, Lifecycle.RUNNING, Lifecycle.RUNNING, true); + } + + private void assertStateEventually(Entity entity, Lifecycle expectedState, Lifecycle state, Boolean isUp) { + Asserts.succeedsEventually(new Runnable() { + @Override public void run() { + assertState(entity, expectedState, state, isUp); + }}); + } + + private void assertState(Entity entity, Lifecycle expectedState, Lifecycle state, Boolean isUp) { + Transition actualExpectedState = entity.sensors().get(Attributes.SERVICE_STATE_EXPECTED); + Lifecycle actualState = entity.sensors().get(Attributes.SERVICE_STATE_ACTUAL); + Boolean actualIsUp = entity.sensors().get(Attributes.SERVICE_UP); + String msg = "actualExpectedState="+actualExpectedState+", actualState="+actualState+", actualIsUp="+actualIsUp; + if (expectedState != null) { + assertEquals(actualExpectedState.getState(), expectedState, msg); + } else { + assertTrue(actualExpectedState == null || actualExpectedState.getState() == null, msg); + } + assertEquals(actualState, state, msg); + assertEquals(actualIsUp, isUp, msg); + } + + private void assertStateEventually(Entity entity, Lifecycle expectedState, Lifecycle state) { + Asserts.succeedsEventually(new Runnable() { + @Override public void run() { + assertState(entity, expectedState, state); + }}); + } + + private void assertState(Entity entity, Lifecycle expectedState, Lifecycle state) { + Transition actualExpectedState = entity.sensors().get(Attributes.SERVICE_STATE_EXPECTED); + Lifecycle actualState = entity.sensors().get(Attributes.SERVICE_STATE_ACTUAL); + Boolean actualIsUp = entity.sensors().get(Attributes.SERVICE_UP); + String msg = "actualExpectedState="+actualExpectedState+", actualState="+actualState+", actualIsUp="+actualIsUp; + if (expectedState != null) { + assertEquals(actualExpectedState.getState(), expectedState, msg); + } else { + assertTrue(actualExpectedState == null || actualExpectedState.getState() == null, msg); + } + assertEquals(actualState, state, msg); + } + + /** + * The AsyncEntity's start leaves it in a "STARTING" state. + * + * It stays like that until {@code clearNotUpIndicator(); setExpected(Lifecycle.RUNNING)} is + * called. It should then report "RUNNING" and service.isUp=true. + */ + @ImplementedBy(AsyncEntityImpl.class) + public interface AsyncEntity extends Entity, Startable { + void setExpected(Lifecycle state); + void addNotUpIndicator(String label, String val); + void clearNotUpIndicators(); + } + + public static class AsyncEntityImpl extends AbstractEntity implements AsyncEntity { + + @Override + public void start(Collection<? extends Location> locations) { + ServiceStateLogic.setExpectedState(this, Lifecycle.STARTING); + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, START.getName(), "starting"); + } + + + @Override + public void setExpected(Lifecycle state) { + ServiceStateLogic.setExpectedState(this, checkNotNull(state, "state")); + } + + @Override + public void clearNotUpIndicators() { + sensors().set(Attributes.SERVICE_NOT_UP_INDICATORS, ImmutableMap.of()); + } + + @Override + public void addNotUpIndicator(String label, String val) { + ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(this, label, val); + } + + @Override + public void stop() { + } + + @Override + public void restart() { + } + } +}
