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 2014/10/01 20:03:02 UTC
[02/15] git commit: set up a new
ManagementContext.getServerExecutionContext() and use that for node-level
tasks, with a tag that can be tracked;
improve lifecycle and configuration for rebind exception handler and for the
persister itself
set up a new ManagementContext.getServerExecutionContext() and use that for node-level tasks, with a tag that can be tracked; improve lifecycle and configuration for rebind exception handler and for the persister itself
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/ea4b4ccb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/ea4b4ccb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/ea4b4ccb
Branch: refs/heads/master
Commit: ea4b4ccb4c26f2c6e7547853e91ff619f2db6365
Parents: 5188c13
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Fri Sep 26 23:27:54 2014 +0100
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Oct 1 16:40:15 2014 +0100
----------------------------------------------------------------------
.../entity/rebind/RebindExceptionHandler.java | 19 ++++-
.../brooklyn/entity/rebind/RebindManager.java | 9 +--
.../brooklyn/management/ExecutionContext.java | 8 ++
.../brooklyn/management/ManagementContext.java | 6 ++
.../mementos/BrooklynMementoPersister.java | 3 +
.../brooklyn/entity/basic/BrooklynTaskTags.java | 4 +
.../java/brooklyn/entity/basic/Entities.java | 2 +-
.../rebind/PeriodicDeltaChangeListener.java | 16 ++--
.../rebind/RebindExceptionHandlerImpl.java | 25 ++++++
.../entity/rebind/RebindManagerImpl.java | 58 +++++++++-----
.../AbstractBrooklynMementoPersister.java | 9 +--
.../BrooklynMementoPersisterToMultiFile.java | 8 ++
.../BrooklynMementoPersisterToObjectStore.java | 81 +++++++++++---------
.../ha/HighAvailabilityManagerImpl.java | 16 +++-
.../internal/AbstractManagementContext.java | 8 +-
.../internal/LocalManagementContext.java | 4 +-
.../NonDeploymentManagementContext.java | 11 +++
.../util/task/AbstractExecutionContext.java | 7 +-
.../util/task/BasicExecutionContext.java | 7 +-
.../entity/rebind/RebindCatalogEntityTest.java | 3 +-
.../entity/rebind/RebindEntityTest.java | 5 +-
.../entity/rebind/RebindFailuresTest.java | 31 +++++---
.../entity/rebind/RebindTestFixture.java | 4 +-
.../brooklyn/entity/rebind/RebindTestUtils.java | 16 +---
.../rebind/RecordingRebindExceptionHandler.java | 5 --
.../transformer/CompoundTransformerTest.java | 1 +
.../HighAvailabilityManagerSplitBrainTest.java | 6 +-
.../ha/HighAvailabilityManagerTestFixture.java | 2 +-
.../brooklyn/management/ha/HotStandbyTest.java | 2 +-
.../AbstractWebAppFixtureIntegrationTest.java | 3 +-
.../brooklyn/launcher/BrooklynLauncher.java | 4 +-
.../BrooklynLauncherHighAvailabilityTest.java | 7 +-
32 files changed, 261 insertions(+), 129 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/api/src/main/java/brooklyn/entity/rebind/RebindExceptionHandler.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/entity/rebind/RebindExceptionHandler.java b/api/src/main/java/brooklyn/entity/rebind/RebindExceptionHandler.java
index 6c80f8b..34c4f05 100644
--- a/api/src/main/java/brooklyn/entity/rebind/RebindExceptionHandler.java
+++ b/api/src/main/java/brooklyn/entity/rebind/RebindExceptionHandler.java
@@ -31,6 +31,14 @@ import com.google.common.annotations.Beta;
/**
* Handler called on all exceptions to do with rebind.
+ * A handler instance is linked to a single rebind pass;
+ * it should not be invoked after {@link #onDone()}.
+ * <p>
+ * {@link #onStart()} must be invoked before the run.
+ * {@link #onDone()} must be invoked after a successful run, and it may throw.
+ * <p>
+ * Implementations may propagate errors or may catch them until {@link #onDone()} is invoked,
+ * and that may throw or report elsewhere, as appropriate.
*
* @author aled
*/
@@ -59,7 +67,6 @@ public interface RebindExceptionHandler {
*/
Enricher onDanglingEnricherRef(String id);
- //XXX
/**
* @return the feed to use in place of the missing one, or null (if hasn't thrown an exception)
*/
@@ -84,7 +91,13 @@ public interface RebindExceptionHandler {
void onManageFailed(BrooklynObjectType type, BrooklynObject instance, Exception e);
- void onDone();
-
+ /** invoked for any high-level, unexpected, or otherwise uncaught failure;
+ * may be invoked on catching above errors */
RuntimeException onFailed(Exception e);
+
+ /** invoked before the rebind pass */
+ void onStart();
+
+ /** invoked after the complete rebind pass, always on success and possibly on failure */
+ void onDone();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/api/src/main/java/brooklyn/entity/rebind/RebindManager.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/entity/rebind/RebindManager.java b/api/src/main/java/brooklyn/entity/rebind/RebindManager.java
index 8bbdcaa..6f15599 100644
--- a/api/src/main/java/brooklyn/entity/rebind/RebindManager.java
+++ b/api/src/main/java/brooklyn/entity/rebind/RebindManager.java
@@ -57,15 +57,14 @@ public interface RebindManager {
@VisibleForTesting
public BrooklynMementoPersister getPersister();
- /**
- * @deprecated since 0.7; use {@link #rebind(ClassLoader)}
- */
- @Deprecated
+ /** @deprecated since 0.7; use {@link #rebind(ClassLoader, RebindExceptionHandler, ManagementNodeState)} */ @Deprecated
public List<Application> rebind() throws IOException;
+ /** @deprecated since 0.7; use {@link #rebind(ClassLoader, RebindExceptionHandler, ManagementNodeState)} */ @Deprecated
public List<Application> rebind(ClassLoader classLoader) throws IOException;
-
+ /** @deprecated since 0.7; use {@link #rebind(ClassLoader, RebindExceptionHandler, ManagementNodeState)} */ @Deprecated
public List<Application> rebind(ClassLoader classLoader, RebindExceptionHandler exceptionHandler) throws IOException;
+ public List<Application> rebind(ClassLoader classLoader, RebindExceptionHandler exceptionHandler, ManagementNodeState mode) throws IOException;
public BrooklynMementoRawData retrieveMementoRawData() throws IOException;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/api/src/main/java/brooklyn/management/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/management/ExecutionContext.java b/api/src/main/java/brooklyn/management/ExecutionContext.java
index 0758ecb..2e43a91 100644
--- a/api/src/main/java/brooklyn/management/ExecutionContext.java
+++ b/api/src/main/java/brooklyn/management/ExecutionContext.java
@@ -48,6 +48,12 @@ public interface ExecutionContext extends Executor {
*/
<T> Task<T> submit(Map<?,?> properties, Callable<T> callable);
+ /** {@link ExecutionManager#submit(Runnable) */
+ Task<?> submit(Runnable runnable);
+
+ /** {@link ExecutionManager#submit(Callable) */
+ <T> Task<T> submit(Callable<T> callable);
+
/** See {@link ExecutionManager#submit(Map, TaskAdaptable)}. */
<T> Task<T> submit(TaskAdaptable<T> task);
@@ -56,4 +62,6 @@ public interface ExecutionContext extends Executor {
*/
<T> Task<T> submit(Map<?,?> properties, TaskAdaptable<T> task);
+ boolean isShutdown();
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/api/src/main/java/brooklyn/management/ManagementContext.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/management/ManagementContext.java b/api/src/main/java/brooklyn/management/ManagementContext.java
index 68722d9..29cb20e 100644
--- a/api/src/main/java/brooklyn/management/ManagementContext.java
+++ b/api/src/main/java/brooklyn/management/ManagementContext.java
@@ -98,6 +98,12 @@ public interface ManagementContext {
* to submit tasks and to observe what tasks are occurring
*/
ExecutionManager getExecutionManager();
+
+ /**
+ * Returns an {@link ExecutionContext} within the {@link ExecutionManager} for tasks
+ * associated to the Brooklyn node's operation (not any entities).
+ **/
+ ExecutionContext getServerExecutionContext();
/**
* Returns the {@link EntityDriverManager} entities can use to create drivers. This
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/api/src/main/java/brooklyn/mementos/BrooklynMementoPersister.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/mementos/BrooklynMementoPersister.java b/api/src/main/java/brooklyn/mementos/BrooklynMementoPersister.java
index 878a211..0317059 100644
--- a/api/src/main/java/brooklyn/mementos/BrooklynMementoPersister.java
+++ b/api/src/main/java/brooklyn/mementos/BrooklynMementoPersister.java
@@ -64,6 +64,9 @@ public interface BrooklynMementoPersister {
void delta(Delta delta, PersistenceExceptionHandler exceptionHandler);
+ void enableWriteAccess();
+ void disableWriteAccess(boolean graceful);
+ /** permanently shuts down all access to the remote store */
void stop(boolean graceful);
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java b/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
index e7a8c70..db16c4b 100644
--- a/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
+++ b/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
@@ -61,7 +61,11 @@ public class BrooklynTaskTags extends TaskTags {
private static final Logger log = LoggerFactory.getLogger(BrooklynTaskTags.WrappedEntity.class);
+ /** Tag for tasks which are running on behalf of the management server, rather than any entity */
+ public static final String BROOKLYN_SERVER_TASK_TAG = "BROOKLYN-SERVER";
+ /** Tag for a task which represents an effector */
public static final String EFFECTOR_TAG = "EFFECTOR";
+ /** Tag for a task which *is* interesting, in contrast to {@link #TRANSIENT_TASK_TAG} */
public static final String NON_TRANSIENT_TASK_TAG = "NON-TRANSIENT";
/** indicates a task is transient, roughly that is to say it is uninteresting --
* specifically this means it can be GC'd as soon as it is completed,
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/basic/Entities.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/basic/Entities.java b/core/src/main/java/brooklyn/entity/basic/Entities.java
index 516fbfa..9a75ab2 100644
--- a/core/src/main/java/brooklyn/entity/basic/Entities.java
+++ b/core/src/main/java/brooklyn/entity/basic/Entities.java
@@ -1007,7 +1007,7 @@ public class Entities {
@Beta
public static ProcessTaskWrapper<Integer> shell(ManagementContext mgmt, String command) {
ProcessTaskWrapper<Integer> t = SystemTasks.exec(command).newTask();
- mgmt.getExecutionManager().submit(t).getUnchecked();
+ mgmt.getServerExecutionContext().submit(t).getUnchecked();
System.out.println(t.getStdout());
System.err.println(t.getStderr());
return t;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/PeriodicDeltaChangeListener.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/PeriodicDeltaChangeListener.java b/core/src/main/java/brooklyn/entity/rebind/PeriodicDeltaChangeListener.java
index 0d708f4..e9cef0c 100644
--- a/core/src/main/java/brooklyn/entity/rebind/PeriodicDeltaChangeListener.java
+++ b/core/src/main/java/brooklyn/entity/rebind/PeriodicDeltaChangeListener.java
@@ -37,6 +37,7 @@ import brooklyn.entity.basic.EntityInternal;
import brooklyn.internal.BrooklynFeatureEnablement;
import brooklyn.location.Location;
import brooklyn.location.basic.LocationInternal;
+import brooklyn.management.ExecutionContext;
import brooklyn.management.ExecutionManager;
import brooklyn.management.Task;
import brooklyn.mementos.BrooklynMementoPersister;
@@ -44,6 +45,7 @@ import brooklyn.policy.Enricher;
import brooklyn.policy.Policy;
import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.exceptions.RuntimeInterruptedException;
+import brooklyn.util.task.BasicExecutionContext;
import brooklyn.util.task.BasicTask;
import brooklyn.util.task.ScheduledTask;
import brooklyn.util.time.Duration;
@@ -93,7 +95,7 @@ public class PeriodicDeltaChangeListener implements ChangeListener {
}
}
- private final ExecutionManager executionManager;
+ private final ExecutionContext executionContext;
private final BrooklynMementoPersister persister;
@@ -117,14 +119,14 @@ public class PeriodicDeltaChangeListener implements ChangeListener {
private final Semaphore persistingMutex = new Semaphore(1);
- /** @deprecated since 0.7.0 pass in a {@link Duration} */
+ /** @deprecated since 0.7.0 pass in an {@link ExecutionContext} and a {@link Duration} */
@Deprecated
public PeriodicDeltaChangeListener(ExecutionManager executionManager, BrooklynMementoPersister persister, PersistenceExceptionHandler exceptionHandler, long periodMillis) {
- this(executionManager, persister, exceptionHandler, Duration.of(periodMillis, TimeUnit.MILLISECONDS));
+ this(new BasicExecutionContext(executionManager), persister, exceptionHandler, Duration.of(periodMillis, TimeUnit.MILLISECONDS));
}
- public PeriodicDeltaChangeListener(ExecutionManager executionManager, BrooklynMementoPersister persister, PersistenceExceptionHandler exceptionHandler, Duration period) {
- this.executionManager = executionManager;
+ public PeriodicDeltaChangeListener(ExecutionContext executionContext, BrooklynMementoPersister persister, PersistenceExceptionHandler exceptionHandler, Duration period) {
+ this.executionContext = executionContext;
this.persister = persister;
this.exceptionHandler = exceptionHandler;
this.period = period;
@@ -165,7 +167,7 @@ public class PeriodicDeltaChangeListener implements ChangeListener {
}});
}
};
- scheduledTask = (ScheduledTask) executionManager.submit(new ScheduledTask(taskFactory).period(period));
+ scheduledTask = (ScheduledTask) executionContext.submit(new ScheduledTask(taskFactory).period(period));
}
void stop() {
@@ -223,7 +225,7 @@ public class PeriodicDeltaChangeListener implements ChangeListener {
* Whether we have been stopped, in which case will not persist or store anything.
*/
private boolean isStopped() {
- return stopped || executionManager.isShutdown();
+ return stopped || executionContext.isShutdown();
}
private void addReferencedObjects(DeltaCollector deltaCollector) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/RebindExceptionHandlerImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/RebindExceptionHandlerImpl.java b/core/src/main/java/brooklyn/entity/rebind/RebindExceptionHandlerImpl.java
index 25523c6..de37f42 100644
--- a/core/src/main/java/brooklyn/entity/rebind/RebindExceptionHandlerImpl.java
+++ b/core/src/main/java/brooklyn/entity/rebind/RebindExceptionHandlerImpl.java
@@ -42,6 +42,7 @@ import brooklyn.util.text.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+/** Stateful handler, meant for a single rebind pass */
public class RebindExceptionHandlerImpl implements RebindExceptionHandler {
private static final Logger LOG = LoggerFactory.getLogger(RebindExceptionHandlerImpl.class);
@@ -62,6 +63,9 @@ public class RebindExceptionHandlerImpl implements RebindExceptionHandler {
protected final Set<Exception> loadPolicyFailures = Sets.newConcurrentHashSet();
protected final List<Exception> exceptions = Collections.synchronizedList(Lists.<Exception>newArrayList());
+ protected boolean started = false;
+ protected boolean done = false;
+
public static Builder builder() {
return new Builder();
}
@@ -99,6 +103,16 @@ public class RebindExceptionHandlerImpl implements RebindExceptionHandler {
this.addPolicyFailureMode = checkNotNull(builder.addPolicyFailureMode, "addPolicyFailureMode");
this.loadPolicyFailureMode = checkNotNull(builder.deserializePolicyFailureMode, "deserializePolicyFailureMode");
}
+
+ public void onStart() {
+ if (done) {
+ throw new IllegalStateException(this+" has already been used on a finished run");
+ }
+ if (started) {
+ throw new IllegalStateException(this+" has already been used on a started run");
+ }
+ started = true;
+ }
@Override
public void onLoadMementoFailed(BrooklynObjectType type, String msg, Exception e) {
@@ -329,6 +343,9 @@ public class RebindExceptionHandlerImpl implements RebindExceptionHandler {
@Override
public RuntimeException onFailed(Exception e) {
+ if (done)
+ throw Exceptions.propagate(e);
+
onDoneImpl(e);
throw new IllegalStateException("Rebind failed", e); // should have thrown exception above
}
@@ -338,9 +355,17 @@ public class RebindExceptionHandlerImpl implements RebindExceptionHandler {
List<Exception> allExceptions = Lists.newArrayList();
+ if (done) {
+ allExceptions.add(new IllegalStateException(this+" has already been informed of rebind done"));
+ }
+ done = true;
+ if (!started) {
+ allExceptions.add(new IllegalStateException(this+" was not informed of start of rebind run"));
+ }
if (e != null) {
allExceptions.add(e);
}
+
if (addPolicyFailureMode != RebindManager.RebindFailureMode.CONTINUE) {
allExceptions.addAll(addPolicyFailures);
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/RebindManagerImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/RebindManagerImpl.java b/core/src/main/java/brooklyn/entity/rebind/RebindManagerImpl.java
index 70a2ba7..47f3770 100644
--- a/core/src/main/java/brooklyn/entity/rebind/RebindManagerImpl.java
+++ b/core/src/main/java/brooklyn/entity/rebind/RebindManagerImpl.java
@@ -57,6 +57,7 @@ import brooklyn.internal.BrooklynFeatureEnablement;
import brooklyn.location.Location;
import brooklyn.location.basic.AbstractLocation;
import brooklyn.location.basic.LocationInternal;
+import brooklyn.management.ExecutionContext;
import brooklyn.management.Task;
import brooklyn.management.ha.HighAvailabilityManagerImpl;
import brooklyn.management.ha.ManagementNodeState;
@@ -127,10 +128,11 @@ public class RebindManagerImpl implements RebindManager {
private volatile boolean persistenceRunning = false;
private volatile PeriodicDeltaChangeListener persistenceRealChangeListener;
private volatile ChangeListener persistencePublicChangeListener;
- private volatile BrooklynMementoPersister persister;
private volatile boolean readOnlyRunning = false;
private volatile ScheduledTask readOnlyTask = null;
+
+ private volatile BrooklynMementoPersister persistenceStoreAccess;
private final boolean persistPoliciesEnabled;
private final boolean persistEnrichersEnabled;
@@ -187,7 +189,7 @@ public class RebindManagerImpl implements RebindManager {
* Must be called before setPerister()
*/
public void setPeriodicPersistPeriod(Duration period) {
- if (persister!=null) throw new IllegalStateException("Cannot set period after persister is generated.");
+ if (persistenceStoreAccess!=null) throw new IllegalStateException("Cannot set period after persister is generated.");
this.periodicPersistPeriod = period;
}
@@ -207,12 +209,12 @@ public class RebindManagerImpl implements RebindManager {
@Override
public void setPersister(BrooklynMementoPersister val, PersistenceExceptionHandler exceptionHandler) {
- if (persister != null && persister != val) {
- throw new IllegalStateException("Dynamically changing persister is not supported: old="+persister+"; new="+val);
+ if (persistenceStoreAccess != null && persistenceStoreAccess != val) {
+ throw new IllegalStateException("Dynamically changing persister is not supported: old="+persistenceStoreAccess+"; new="+val);
}
- this.persister = checkNotNull(val, "persister");
+ this.persistenceStoreAccess = checkNotNull(val, "persister");
- this.persistenceRealChangeListener = new PeriodicDeltaChangeListener(managementContext.getExecutionManager(), persister, exceptionHandler, periodicPersistPeriod);
+ this.persistenceRealChangeListener = new PeriodicDeltaChangeListener(managementContext.getServerExecutionContext(), persistenceStoreAccess, exceptionHandler, periodicPersistPeriod);
this.persistencePublicChangeListener = new SafeChangeListener(persistenceRealChangeListener);
if (persistenceRunning) {
@@ -223,7 +225,7 @@ public class RebindManagerImpl implements RebindManager {
@Override
@VisibleForTesting
public BrooklynMementoPersister getPersister() {
- return persister;
+ return persistenceStoreAccess;
}
@Override
@@ -233,6 +235,7 @@ public class RebindManagerImpl implements RebindManager {
}
LOG.debug("Starting persistence, mgmt "+managementContext.getManagementNodeId());
persistenceRunning = true;
+ persistenceStoreAccess.enableWriteAccess();
if (persistenceRealChangeListener != null) persistenceRealChangeListener.start();
}
@@ -241,7 +244,7 @@ public class RebindManagerImpl implements RebindManager {
LOG.debug("Stopping persistence, mgmt "+managementContext.getManagementNodeId());
persistenceRunning = false;
if (persistenceRealChangeListener != null) persistenceRealChangeListener.stop();
- if (persister != null) persister.stop(true);
+ if (persistenceStoreAccess != null) persistenceStoreAccess.disableWriteAccess(true);
LOG.debug("Stopped persistence, mgmt "+managementContext.getManagementNodeId());
}
@@ -294,8 +297,8 @@ public class RebindManagerImpl implements RebindManager {
}});
}
};
- readOnlyTask = (ScheduledTask) managementContext.getExecutionManager().submit(
- new ScheduledTask(taskFactory).period(periodicPersistPeriod));
+ readOnlyTask = (ScheduledTask) managementContext.getServerExecutionContext().submit(
+ new ScheduledTask(MutableMap.of("displayName", "Periodic read-only rebind"), taskFactory).period(periodicPersistPeriod));
}
@Override
@@ -326,6 +329,7 @@ public class RebindManagerImpl implements RebindManager {
public void stop() {
stopReadOnly();
stopPersistence();
+ if (persistenceStoreAccess != null) persistenceStoreAccess.stop(true);
}
protected ManagementNodeState getRebindMode() {
@@ -344,9 +348,9 @@ public class RebindManagerImpl implements RebindManager {
@Override
@VisibleForTesting
public void waitForPendingComplete(Duration timeout) throws InterruptedException, TimeoutException {
- if (persister == null || !persistenceRunning) return;
+ if (persistenceStoreAccess == null || !persistenceRunning) return;
persistenceRealChangeListener.waitForPendingComplete(timeout);
- persister.waitForWritesCompleted(timeout);
+ persistenceStoreAccess.waitForWritesCompleted(timeout);
}
@Override
@VisibleForTesting
@@ -366,7 +370,7 @@ public class RebindManagerImpl implements RebindManager {
@Override
public List<Application> rebind(final ClassLoader classLoader) throws IOException {
- return rebind(classLoader, null);
+ return rebind(classLoader, null, null);
}
@Override
@@ -374,7 +378,8 @@ public class RebindManagerImpl implements RebindManager {
return rebind(classLoader, exceptionHandler, null);
}
- protected List<Application> rebind(ClassLoader classLoaderO, RebindExceptionHandler exceptionHandlerO, ManagementNodeState modeO) throws IOException {
+ @Override
+ public List<Application> rebind(ClassLoader classLoaderO, RebindExceptionHandler exceptionHandlerO, ManagementNodeState modeO) throws IOException {
final ClassLoader classLoader = classLoaderO!=null ? classLoaderO :
managementContext.getCatalog().getRootClassLoader();
final RebindExceptionHandler exceptionHandler = exceptionHandlerO!=null ? exceptionHandlerO :
@@ -386,9 +391,12 @@ public class RebindManagerImpl implements RebindManager {
.build();
final ManagementNodeState mode = modeO!=null ? modeO : getRebindMode();
- BasicExecutionContext ec = BasicExecutionContext.getCurrentExecutionContext();
+ if (mode!=ManagementNodeState.HOT_STANDBY && mode!=ManagementNodeState.MASTER)
+ throw new IllegalStateException("Must be either master or read only to rebind (mode "+mode+")");
+
+ ExecutionContext ec = BasicExecutionContext.getCurrentExecutionContext();
if (ec == null) {
- ec = new BasicExecutionContext(managementContext.getExecutionManager());
+ ec = managementContext.getServerExecutionContext();
Task<List<Application>> task = ec.submit(new Callable<List<Application>>() {
@Override public List<Application> call() throws Exception {
return rebindImpl(classLoader, exceptionHandler, mode);
@@ -422,11 +430,11 @@ public class RebindManagerImpl implements RebindManager {
*/
protected BrooklynMementoRawData loadMementoRawData(final RebindExceptionHandler exceptionHandler) throws IOException {
try {
- if (!(persister instanceof BrooklynMementoPersisterToObjectStore)) {
- throw new IllegalStateException("Cannot load raw memento with persister "+persister);
+ if (!(persistenceStoreAccess instanceof BrooklynMementoPersisterToObjectStore)) {
+ throw new IllegalStateException("Cannot load raw memento with persister "+persistenceStoreAccess);
}
- return ((BrooklynMementoPersisterToObjectStore)persister).loadMementoRawData(exceptionHandler);
+ return ((BrooklynMementoPersisterToObjectStore)persistenceStoreAccess).loadMementoRawData(exceptionHandler);
} catch (RuntimeException e) {
throw exceptionHandler.onFailed(e);
@@ -440,6 +448,8 @@ public class RebindManagerImpl implements RebindManager {
RebindTracker.setRebinding();
try {
+ exceptionHandler.onStart();
+
Reflections reflections = new Reflections(classLoader);
RebindContextImpl rebindContext = new RebindContextImpl(exceptionHandler, classLoader);
LookupContext realLookupContext = new RebindContextLookupContext(managementContext, rebindContext, exceptionHandler);
@@ -463,7 +473,7 @@ public class RebindManagerImpl implements RebindManager {
// PHASE ONE
//
- BrooklynMementoManifest mementoManifest = persister.loadMementoManifest(exceptionHandler);
+ BrooklynMementoManifest mementoManifest = persistenceStoreAccess.loadMementoManifest(exceptionHandler);
boolean isEmpty = mementoManifest.isEmpty();
if (!isEmpty) {
@@ -512,7 +522,7 @@ public class RebindManagerImpl implements RebindManager {
// PHASE THREE
//
- BrooklynMemento memento = persister.loadMemento(realLookupContext, exceptionHandler);
+ BrooklynMemento memento = persistenceStoreAccess.loadMemento(realLookupContext, exceptionHandler);
//
@@ -786,6 +796,7 @@ public class RebindManagerImpl implements RebindManager {
managementContext.getCatalog().reset(rebindContext.getCatalogItems());
} else if (shouldLoadDefaultCatalog) {
// Load catalogue as normal
+ // TODO in read-only mode, should do this less frequently than entities etc
LOG.debug("RebindManager loading default catalog");
((BasicBrooklynCatalog) managementContext.getCatalog()).resetCatalogToContentsAtConfiguredUrl();
} else {
@@ -1080,4 +1091,9 @@ public class RebindManagerImpl implements RebindManager {
}
}
}
+
+ @Override
+ public String toString() {
+ return super.toString()+"[mgmt="+managementContext.getManagementNodeId()+"]";
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/persister/AbstractBrooklynMementoPersister.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/AbstractBrooklynMementoPersister.java b/core/src/main/java/brooklyn/entity/rebind/persister/AbstractBrooklynMementoPersister.java
index 32bfcc3..c230e9e 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/AbstractBrooklynMementoPersister.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/AbstractBrooklynMementoPersister.java
@@ -67,11 +67,10 @@ public abstract class AbstractBrooklynMementoPersister implements BrooklynMement
}
return builder.build();
}
-
- @Override
- public void stop(boolean graceful) {
- // no-op
- }
+
+ @Override public void enableWriteAccess() {}
+ @Override public void disableWriteAccess(boolean graceful) {}
+ @Override public void stop(boolean graceful) {}
@Override
public void checkpoint(BrooklynMemento newMemento, PersistenceExceptionHandler exceptionHandler) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToMultiFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToMultiFile.java b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToMultiFile.java
index ab0f831..aeedf61 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToMultiFile.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToMultiFile.java
@@ -128,6 +128,14 @@ public class BrooklynMementoPersisterToMultiFile implements BrooklynMementoPersi
LOG.info("Memento-persister will use directory {}", dir);
}
+ @Override public void enableWriteAccess() {
+ // no-op -- means subsequent writes will not be enabled
+ }
+
+ @Override public void disableWriteAccess(boolean graceful) {
+ stop(graceful);
+ }
+
@Override
public void stop(boolean graceful) {
running = false;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
index 5b224b1..e02de72 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
@@ -101,8 +101,8 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
private final ListeningExecutorService executor;
- private volatile boolean running = true;
- private volatile boolean shuttingDown = false;
+ private volatile boolean writesAllowed = false;
+ private volatile boolean writesShuttingDown = false;
/**
* Lock used on writes (checkpoint + delta) so that {@link #waitForWritesCompleted(Duration)} can block
@@ -135,30 +135,45 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
}}));
}
+ @Override public void enableWriteAccess() {
+ writesAllowed = true;
+ }
+
@Override
- public void stop(boolean graceful) {
- shuttingDown = true;
+ public void disableWriteAccess(boolean graceful) {
+ writesShuttingDown = true;
try {
- running = false;
- if (executor != null) {
- if (graceful) {
- // a very long timeout to ensure we don't lose state.
- // If persisting thousands of entities over slow network to Object Store, could take minutes.
- executor.shutdown();
- try {
- executor.awaitTermination(1, TimeUnit.HOURS);
- } catch (InterruptedException e) {
- throw Exceptions.propagate(e);
- }
- } else {
- executor.shutdownNow();
- }
- }
+ writesAllowed = false;
+ // a very long timeout to ensure we don't lose state.
+ // If persisting thousands of entities over slow network to Object Store, could take minutes.
+ waitForWritesCompleted(Duration.ONE_HOUR);
+
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
} finally {
- shuttingDown = false;
+ writesShuttingDown = false;
}
}
+ @Override
+ public void stop(boolean graceful) {
+ disableWriteAccess(graceful);
+
+ if (executor != null) {
+ if (graceful) {
+ executor.shutdown();
+ try {
+ // should be quick because we've just turned off writes, waiting for their completion
+ executor.awaitTermination(1, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ throw Exceptions.propagate(e);
+ }
+ } else {
+ executor.shutdownNow();
+ }
+ }
+ }
+
public PersistenceObjectStore getObjectStore() {
return objectStore;
}
@@ -331,10 +346,6 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
public void visit(String contents, BrooklynObjectType type, String subPath) throws Exception;
}
protected void visitMemento(final Visitor visitor, final RebindExceptionHandler exceptionHandler) throws IOException {
- if (!running && !shuttingDown) {
- throw new IllegalStateException("Persister not running; cannot load memento manifest from " + objectStore.getSummaryName());
- }
-
List<String> entitySubPathList;
List<String> locationSubPathList;
List<String> policySubPathList;
@@ -429,12 +440,15 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
}
}
+ protected void checkWritesAllowed() {
+ if (!writesAllowed && !writesShuttingDown) {
+ throw new IllegalStateException("Writes not allowed in "+this);
+ }
+ }
+
@Beta
public void checkpoint(BrooklynMementoRawData newMemento, PersistenceExceptionHandler exceptionHandler) {
- if (!running && !shuttingDown) {
- if (LOG.isDebugEnabled()) LOG.debug("Ignoring checkpointing entire memento, because not running");
- return;
- }
+ checkWritesAllowed();
try {
lock.writeLock().lockInterruptibly();
@@ -486,10 +500,7 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
@Override
public void checkpoint(BrooklynMemento newMemento, PersistenceExceptionHandler exceptionHandler) {
- if (!running && !shuttingDown) {
- if (LOG.isDebugEnabled()) LOG.debug("Ignoring checkpointing entire memento, because not running");
- return;
- }
+ checkWritesAllowed();
Delta delta = PersisterDeltaImpl.builder()
.entities(newMemento.getEntityMementos().values())
@@ -506,10 +517,8 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
@Override
public void delta(Delta delta, PersistenceExceptionHandler exceptionHandler) {
- if (!running && !shuttingDown) {
- if (LOG.isDebugEnabled()) LOG.debug("Ignoring checkpointed delta of memento, because not running");
- return;
- }
+ checkWritesAllowed();
+
Stopwatch stopwatch = deltaImpl(delta, exceptionHandler);
if (LOG.isDebugEnabled()) LOG.debug("Checkpointed delta of memento in {}; updated {} entities, {} locations, " +
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java b/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
index d63aa99..52fffdc 100644
--- a/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
+++ b/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
@@ -74,7 +74,7 @@ import com.google.common.collect.Iterables;
* Promotion to master involves:
* <ol>
* <li>notifying the other management-nodes that it is now master
- * <li>calling {@link RebindManager#rebind()} to read all persisted entity state, and thus reconstitute the entities.
+ * <li>calling {@link RebindManager#rebind(ClassLoader, brooklyn.entity.rebind.RebindExceptionHandler, ManagementNodeState)} to read all persisted entity state, and thus reconstitute the entities.
* </ol>
* <p>
* Future improvements in this area will include brooklyn-managing-brooklyn to decide + promote
@@ -212,6 +212,10 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
// so another node could have already become master or terminated in that window.
ManagementNodeSyncRecord existingMaster = hasHealthyMaster();
+ // catch error in some tests where mgmt context has a different mgmt context
+ if (managementContext.getHighAvailabilityManager()!=this)
+ throw new IllegalStateException("Cannot start an HA manager on a management context with a different HA manager!");
+
switch (startMode) {
case AUTO:
// don't care; let's start and see if we promote ourselves
@@ -531,10 +535,16 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
LOG.warn("Problem in promption-listener (continuing)", e);
}
}
+ boolean wasHotStandby = nodeState==ManagementNodeState.HOT_STANDBY;
nodeState = ManagementNodeState.MASTER;
publishPromotionToMaster();
try {
- managementContext.getRebindManager().rebind(managementContext.getCatalog().getRootClassLoader());
+ if (wasHotStandby) {
+ // could just promote the standby items; but for now we stop the old read-only and re-load them, to make sure nothing has been missed
+ // TODO ideally there'd be an incremental rebind as well as an incremental persist
+ managementContext.getRebindManager().stopReadOnly();
+ }
+ managementContext.getRebindManager().rebind(managementContext.getCatalog().getRootClassLoader(), null, nodeState);
} catch (Exception e) {
LOG.error("Management node enountered problem during rebind when promoting self to master; demoting to FAILED and rethrowing: "+e);
demoteToFailed();
@@ -568,7 +578,7 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
}
protected void onDemotionStopTasks() {
- managementContext.getRebindManager().stop();
+ managementContext.getRebindManager().stopPersistence();
for (Application app: managementContext.getApplications())
Entities.unmanage(app);
// let's try forcibly interrupting tasks on managed entities
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/management/internal/AbstractManagementContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/AbstractManagementContext.java b/core/src/main/java/brooklyn/management/internal/AbstractManagementContext.java
index 44a77ae..b91e99b 100644
--- a/core/src/main/java/brooklyn/management/internal/AbstractManagementContext.java
+++ b/core/src/main/java/brooklyn/management/internal/AbstractManagementContext.java
@@ -212,7 +212,13 @@ public abstract class AbstractManagementContext implements ManagementContextInte
// BEC is a thin wrapper around EM so fine to create a new one here
return new BasicExecutionContext(MutableMap.of("tag", BrooklynTaskTags.tagForContextEntity(e)), getExecutionManager());
}
-
+
+ @Override
+ public ExecutionContext getServerExecutionContext() {
+ // BEC is a thin wrapper around EM so fine to create a new one here
+ return new BasicExecutionContext(MutableMap.of("tag", BrooklynTaskTags.BROOKLYN_SERVER_TASK_TAG), getExecutionManager());
+ }
+
@Override
public SubscriptionContext getSubscriptionContext(Entity e) {
// BSC is a thin wrapper around SM so fine to create a new one here
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java b/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
index 2c3d43a..bf4a559 100644
--- a/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
+++ b/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
@@ -39,6 +39,7 @@ import brooklyn.config.BrooklynProperties.Factory.Builder;
import brooklyn.entity.Application;
import brooklyn.entity.Effector;
import brooklyn.entity.Entity;
+import brooklyn.entity.basic.BrooklynTaskTags;
import brooklyn.entity.drivers.downloads.BasicDownloadsManager;
import brooklyn.entity.effector.Effectors;
import brooklyn.entity.proxying.InternalEntityFactory;
@@ -54,6 +55,7 @@ import brooklyn.management.SubscriptionManager;
import brooklyn.management.Task;
import brooklyn.management.TaskAdaptable;
import brooklyn.management.ha.OsgiManager;
+import brooklyn.util.collections.MutableMap;
import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.guava.Maybe;
import brooklyn.util.task.BasicExecutionContext;
@@ -309,7 +311,7 @@ public class LocalManagementContext extends AbstractManagementContext {
}
return execution;
}
-
+
@Override
public void terminate() {
INSTANCES.remove(this);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java b/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
index a404bb8..21f1968 100644
--- a/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
+++ b/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
@@ -20,6 +20,7 @@ package brooklyn.management.internal;
import static com.google.common.base.Preconditions.checkNotNull;
+import java.io.IOException;
import java.net.URI;
import java.net.URL;
import java.util.Collection;
@@ -231,6 +232,11 @@ public class NonDeploymentManagementContext implements ManagementContextInternal
return initialManagementContext.getExecutionContext(entity);
}
+ @Override
+ public ExecutionContext getServerExecutionContext() {
+ return initialManagementContext.getServerExecutionContext();
+ }
+
// TODO the methods below should delegate to the application?
@Override
public EntityDriverManager getEntityDriverManager() {
@@ -445,6 +451,11 @@ public class NonDeploymentManagementContext implements ManagementContextInternal
public List<Application> rebind(ClassLoader classLoader, RebindExceptionHandler exceptionHandler) {
throw new IllegalStateException("Non-deployment context "+NonDeploymentManagementContext.this+" is not valid for this operation.");
}
+
+ @Override
+ public List<Application> rebind(ClassLoader classLoader, RebindExceptionHandler exceptionHandler, ManagementNodeState mode) throws IOException {
+ throw new IllegalStateException("Non-deployment context "+NonDeploymentManagementContext.this+" is not valid for this operation.");
+ }
@Override
public void startPersistence() {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/util/task/AbstractExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/util/task/AbstractExecutionContext.java b/core/src/main/java/brooklyn/util/task/AbstractExecutionContext.java
index cd94915..3002922 100644
--- a/core/src/main/java/brooklyn/util/task/AbstractExecutionContext.java
+++ b/core/src/main/java/brooklyn/util/task/AbstractExecutionContext.java
@@ -37,21 +37,27 @@ public abstract class AbstractExecutionContext implements ExecutionContext {
*
* @see ExecutionManager#submit(Map, Task)
*/
+ @Override
public Task<?> submit(Map<?,?> properties, Runnable runnable) { return submitInternal(properties, runnable); }
/** @see #submit(Map, Runnable) */
+ @Override
public Task<?> submit(Runnable runnable) { return submitInternal(Maps.newLinkedHashMap(), runnable); }
/** @see #submit(Map, Runnable) */
+ @Override
public <T> Task<T> submit(Callable<T> callable) { return submitInternal(Maps.newLinkedHashMap(), callable); }
/** @see #submit(Map, Runnable) */
+ @Override
public <T> Task<T> submit(Map<?,?> properties, Callable<T> callable) { return submitInternal(properties, callable); }
/** @see #submit(Map, Runnable) */
+ @Override
public <T> Task<T> submit(TaskAdaptable<T> task) { return submitInternal(Maps.newLinkedHashMap(), task.asTask()); }
/** @see #submit(Map, Runnable) */
+ @Override
public <T> Task<T> submit(Map<?,?> properties, TaskAdaptable<T> task) { return submitInternal(properties, task.asTask()); }
/**
@@ -66,5 +72,4 @@ public abstract class AbstractExecutionContext implements ExecutionContext {
/** does the work internally of submitting the task; note that the return value may be a wrapper task even if a task is passed in,
* if the execution context where the target should run is different (e.g. submitting an effector task cross-context) */
protected abstract <T> Task<T> submitInternal(Map<?,?> properties, Object task);
-
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/main/java/brooklyn/util/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/util/task/BasicExecutionContext.java b/core/src/main/java/brooklyn/util/task/BasicExecutionContext.java
index a7bcbbd..ee1818e 100644
--- a/core/src/main/java/brooklyn/util/task/BasicExecutionContext.java
+++ b/core/src/main/java/brooklyn/util/task/BasicExecutionContext.java
@@ -190,7 +190,12 @@ public class BasicExecutionContext extends AbstractExecutionContext {
private void registerPerThreadExecutionContext() { perThreadExecutionContext.set(this); }
private void clearPerThreadExecutionContext() { perThreadExecutionContext.remove(); }
-
+
+ @Override
+ public boolean isShutdown() {
+ return getExecutionManager().isShutdown();
+ }
+
@Override
public String toString() {
return super.toString()+"("+tags+")";
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RebindCatalogEntityTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindCatalogEntityTest.java b/core/src/test/java/brooklyn/entity/rebind/RebindCatalogEntityTest.java
index e482577..eac2e35 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindCatalogEntityTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindCatalogEntityTest.java
@@ -39,6 +39,7 @@ import brooklyn.entity.basic.EntityInternal;
import brooklyn.entity.basic.StartableApplication;
import brooklyn.entity.proxying.EntitySpec;
import brooklyn.event.basic.Sensors;
+import brooklyn.management.ha.ManagementNodeState;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.util.javalang.UrlClassLoader;
@@ -128,7 +129,7 @@ public class RebindCatalogEntityTest extends RebindTestFixture<StartableApplicat
newManagementContext.getCatalog().addItem(appClazz);
ClassLoader classLoader = newManagementContext.getCatalog().getRootClassLoader();
- List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader);
+ List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader, null, ManagementNodeState.MASTER);
newManagementContext.getRebindManager().startPersistence();
return (StartableApplication) newApps.get(0);
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java b/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
index fed6dd9..3450c40 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
@@ -61,6 +61,7 @@ import brooklyn.event.basic.DependentConfiguration;
import brooklyn.event.basic.Sensors;
import brooklyn.location.Location;
import brooklyn.location.basic.LocationConfigTest.MyLocation;
+import brooklyn.management.ha.ManagementNodeState;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.mementos.EntityMemento;
import brooklyn.test.Asserts;
@@ -338,7 +339,7 @@ public class RebindEntityTest extends RebindTestFixtureWithApp {
Thread thread = new Thread() {
public void run() {
try {
- newManagementContext.getRebindManager().rebind(classLoader);
+ newManagementContext.getRebindManager().rebind(classLoader, null, ManagementNodeState.MASTER);
} catch (Exception e) {
throw Throwables.propagate(e);
}
@@ -578,7 +579,7 @@ public class RebindEntityTest extends RebindTestFixtureWithApp {
RebindTestUtils.waitForPersisted(origManagementContext);
newManagementContext = RebindTestUtils.newPersistingManagementContextUnstarted(mementoDir, classLoader);
- List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader);
+ List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader, null, ManagementNodeState.MASTER);
newManagementContext.getRebindManager().startPersistence();
assertEquals(newApps.size(), 0, "apps="+newApps);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RebindFailuresTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindFailuresTest.java b/core/src/test/java/brooklyn/entity/rebind/RebindFailuresTest.java
index 9714dc9..e070bd0 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindFailuresTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindFailuresTest.java
@@ -27,6 +27,7 @@ import java.io.File;
import java.util.Map;
import java.util.Set;
+import org.testng.Assert;
import org.testng.annotations.Test;
import brooklyn.config.ConfigKey;
@@ -43,12 +44,13 @@ import brooklyn.entity.rebind.RebindManager.RebindFailureMode;
import brooklyn.entity.trait.Identifiable;
import brooklyn.event.AttributeSensor;
import brooklyn.management.EntityManager;
-import brooklyn.management.internal.LocalManagementContext;
import brooklyn.policy.Enricher;
import brooklyn.policy.EnricherSpec;
import brooklyn.policy.Policy;
import brooklyn.policy.PolicySpec;
import brooklyn.policy.basic.AbstractPolicy;
+import brooklyn.test.entity.LocalManagementContextForTests;
+import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.flags.SetFromFlag;
import brooklyn.util.os.Os;
@@ -91,14 +93,15 @@ public class RebindFailuresTest extends RebindTestFixtureWithApp {
.impl(MyEntityFailingImpl.class)
.configure(MyEntityFailingImpl.FAIL_ON_REBIND, true));
- newManagementContext = new LocalManagementContext();
+ newManagementContext = LocalManagementContextForTests.newInstance();
EntityManager newEntityManager = newManagementContext.getEntityManager();
RecordingRebindExceptionHandler exceptionHandler = new RecordingRebindExceptionHandler(danglingRefFailureMode, rebindFailureMode);
try {
newApp = rebind(newManagementContext, exceptionHandler);
fail();
} catch (Exception e) {
- if (!e.toString().contains("Problems rebinding")) throw e; // expected
+ assertFailureRebindingError(e);
+ Assert.assertTrue(e.toString().toLowerCase().contains("rebinding entity"), "Wrong error: "+e);
}
// exception handler should have been told about failure
@@ -117,14 +120,15 @@ public class RebindFailuresTest extends RebindTestFixtureWithApp {
.impl(MyEntityFailingImpl.class)
.configure(MyEntityFailingImpl.FAIL_ON_REBIND, true));
- newManagementContext = new LocalManagementContext();
+ newManagementContext = LocalManagementContextForTests.newInstance();
EntityManager newEntityManager = newManagementContext.getEntityManager();
RecordingRebindExceptionHandler exceptionHandler = new RecordingRebindExceptionHandler(danglingRefFailureMode, rebindFailureMode);
try {
newApp = rebind(newManagementContext, exceptionHandler);
fail();
} catch (Exception e) {
- if (!e.toString().contains("Problems rebinding")) throw e; // expected
+ assertFailureRebindingError(e);
+ Assert.assertTrue(e.toString().toLowerCase().contains("rebinding entity"), "Wrong error: "+e);
}
// exception handler should have been told about failure
@@ -143,7 +147,7 @@ public class RebindFailuresTest extends RebindTestFixtureWithApp {
.impl(MyEntityFailingImpl.class)
.configure(MyEntityFailingImpl.FAIL_ON_REBIND, true));
- newManagementContext = new LocalManagementContext();
+ newManagementContext = LocalManagementContextForTests.newInstance();
EntityManager newEntityManager = newManagementContext.getEntityManager();
RecordingRebindExceptionHandler exceptionHandler = new RecordingRebindExceptionHandler(danglingRefFailureMode, rebindFailureMode);
newApp = rebind(newManagementContext, exceptionHandler);
@@ -165,16 +169,13 @@ public class RebindFailuresTest extends RebindTestFixtureWithApp {
File entitiesDir = Os.mkdirs(new File(mementoDir, "entities"));
Files.write("invalid text", new File(entitiesDir, "mycorruptfile"), Charsets.UTF_8);
- newManagementContext = new LocalManagementContext();
+ newManagementContext = LocalManagementContextForTests.newInstance();
RecordingRebindExceptionHandler exceptionHandler = new RecordingRebindExceptionHandler(danglingRefFailureMode, rebindFailureMode);
try {
newApp = rebind(newManagementContext, exceptionHandler);
fail();
} catch (Exception e) {
- if (e.toString().contains("Problem rebinding") || e.toString().contains("Problems rebinding"))
- /* expected */ ;
- else
- throw e;
+ assertFailureRebindingError(e);
}
// exception handler should have been told about failure
@@ -182,6 +183,14 @@ public class RebindFailuresTest extends RebindTestFixtureWithApp {
assertEquals(exceptionHandler.loadMementoFailures.size(), 2, "exceptions="+exceptionHandler.loadMementoFailures);
}
+ protected void assertFailureRebindingError(Exception e) {
+ if (e.toString().toLowerCase().matches(".*(problem|failure)(s?) rebinding.*")) {
+ // expected
+ } else {
+ throw Exceptions.propagate(e);
+ }
+ }
+
@Test
public void testRebindWithFailingPolicyContinuesWithoutPolicy() throws Exception {
origApp.addPolicy(PolicySpec.create(MyPolicyFailingImpl.class)
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java b/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
index f487252..f2314f7 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
@@ -94,7 +94,7 @@ public abstract class RebindTestFixture<T extends StartableApplication> {
}
protected boolean useEmptyCatalog() {
- return false;
+ return true;
}
protected int getPersistPeriodMillis() {
@@ -158,7 +158,7 @@ public abstract class RebindTestFixture<T extends StartableApplication> {
}
protected BrooklynMementoManifest loadMementoManifest() throws Exception {
- newManagementContext = new LocalManagementContextForTests();
+ newManagementContext = createNewManagementContext();
FileBasedObjectStore objectStore = new FileBasedObjectStore(mementoDir);
objectStore.injectManagementContext(newManagementContext);
objectStore.prepareForSharedUse(PersistMode.AUTO, HighAvailabilityMode.DISABLED);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java b/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
index 65738a7..ee34e54 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
@@ -45,6 +45,7 @@ import brooklyn.entity.trait.Identifiable;
import brooklyn.location.Location;
import brooklyn.management.ManagementContext;
import brooklyn.management.ha.HighAvailabilityMode;
+import brooklyn.management.ha.ManagementNodeState;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.management.internal.ManagementContextInternal;
import brooklyn.mementos.BrooklynMemento;
@@ -249,12 +250,8 @@ public class RebindTestUtils {
public static Collection<Application> rebindAll(LocalManagementContext newManagementContext, ClassLoader classLoader, RebindExceptionHandler exceptionHandler) throws Exception {
LOG.info("Rebinding app");
- List<Application> newApps;
- if (exceptionHandler == null) {
- newApps = newManagementContext.getRebindManager().rebind(classLoader);
- } else {
- newApps = newManagementContext.getRebindManager().rebind(classLoader, exceptionHandler);
- }
+ List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader, exceptionHandler, ManagementNodeState.MASTER);
+
if (newApps.isEmpty()) throw new IllegalStateException("Application could not be rebinded; serialization probably failed");
newManagementContext.getRebindManager().startPersistence();
return newApps;
@@ -286,12 +283,7 @@ public class RebindTestUtils {
((ManagementContextInternal)newManagementContext).getBrooklynProperties(),
classLoader);
newManagementContext.getRebindManager().setPersister(newPersister, PersistenceExceptionHandlerImpl.builder().build());
- List<Application> newApps;
- if (exceptionHandler == null) {
- newApps = newManagementContext.getRebindManager().rebind(classLoader);
- } else {
- newApps = newManagementContext.getRebindManager().rebind(classLoader, exceptionHandler);
- }
+ List<Application> newApps = newManagementContext.getRebindManager().rebind(classLoader, exceptionHandler, ManagementNodeState.MASTER);
newManagementContext.getRebindManager().startPersistence();
return newApps;
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/RecordingRebindExceptionHandler.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RecordingRebindExceptionHandler.java b/core/src/test/java/brooklyn/entity/rebind/RecordingRebindExceptionHandler.java
index 2cec642..d2c8b8c 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RecordingRebindExceptionHandler.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RecordingRebindExceptionHandler.java
@@ -82,11 +82,6 @@ public class RecordingRebindExceptionHandler extends RebindExceptionHandlerImpl
}
@Override
- public void onDone() {
- super.onDone();
- }
-
- @Override
public RuntimeException onFailed(Exception e) {
failed = e;
return super.onFailed(e);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/entity/rebind/transformer/CompoundTransformerTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/transformer/CompoundTransformerTest.java b/core/src/test/java/brooklyn/entity/rebind/transformer/CompoundTransformerTest.java
index 6e9ab65..9474cd8 100644
--- a/core/src/test/java/brooklyn/entity/rebind/transformer/CompoundTransformerTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/transformer/CompoundTransformerTest.java
@@ -141,6 +141,7 @@ public class CompoundTransformerTest extends RebindTestFixtureWithApp {
objectStore,
((ManagementContextInternal)origManagementContext).getBrooklynProperties(),
origManagementContext.getCatalog().getRootClassLoader());
+ persister.enableWriteAccess();
PersistenceExceptionHandler exceptionHandler = PersistenceExceptionHandlerImpl.builder().build();
persister.checkpoint(rawData, exceptionHandler);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerSplitBrainTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerSplitBrainTest.java b/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerSplitBrainTest.java
index dbdf3d3..f7399d4 100644
--- a/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerSplitBrainTest.java
+++ b/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerSplitBrainTest.java
@@ -101,7 +101,7 @@ public class HighAvailabilityManagerSplitBrainTest {
((ManagementPlaneSyncRecordPersisterToObjectStore)persister).allowRemoteTimestampInMemento();
BrooklynMementoPersisterToObjectStore persisterObj = new BrooklynMementoPersisterToObjectStore(objectStore, mgmt.getBrooklynProperties(), classLoader);
mgmt.getRebindManager().setPersister(persisterObj, PersistenceExceptionHandlerImpl.builder().build());
- ha = new HighAvailabilityManagerImpl(mgmt)
+ ha = ((HighAvailabilityManagerImpl)mgmt.getHighAvailabilityManager())
.setPollPeriod(Duration.PRACTICALLY_FOREVER)
.setHeartbeatTimeout(Duration.THIRTY_SECONDS)
.setLocalTicker(ticker)
@@ -270,9 +270,9 @@ public class HighAvailabilityManagerSplitBrainTest {
assertEquals(memento2c.getManagementNodes().get(n1.ownNodeId).getRemoteTimestamp(), time2);
assertEquals(memento2c.getManagementNodes().get(n2.ownNodeId).getRemoteTimestamp(), time2);
- // and no entities at n1
- assertEquals(n1.mgmt.getApplications().size(), 0);
+ // right number of entities at n2; n1 may or may not depending whether hot standby is default
assertEquals(n2.mgmt.getApplications().size(), 1);
+// assertEquals(n1.mgmt.getApplications().size(), 0);
}
@Test(invocationCount=50, groups="Integration")
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java b/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
index a657636..c646f69 100644
--- a/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
+++ b/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
@@ -91,7 +91,7 @@ public abstract class HighAvailabilityManagerTestFixture {
managementContext.getBrooklynProperties(),
classLoader);
managementContext.getRebindManager().setPersister(persisterObj, PersistenceExceptionHandlerImpl.builder().build());
- manager = new HighAvailabilityManagerImpl(managementContext)
+ manager = ((HighAvailabilityManagerImpl)managementContext.getHighAvailabilityManager())
.setPollPeriod(getPollPeriod())
.setHeartbeatTimeout(Duration.THIRTY_SECONDS)
.setPromotionListener(promotionListener)
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/core/src/test/java/brooklyn/management/ha/HotStandbyTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/management/ha/HotStandbyTest.java b/core/src/test/java/brooklyn/management/ha/HotStandbyTest.java
index e4219f7..cc9f065 100644
--- a/core/src/test/java/brooklyn/management/ha/HotStandbyTest.java
+++ b/core/src/test/java/brooklyn/management/ha/HotStandbyTest.java
@@ -81,7 +81,7 @@ public class HotStandbyTest {
.setPollPeriod(Duration.PRACTICALLY_FOREVER)
.setHeartbeatTimeout(Duration.THIRTY_SECONDS)
.setPersister(persister);
- log.info("Created "+nodeName+" "+ownNodeId+" - RM "+mgmt.getRebindManager());
+ log.info("Created "+nodeName+" "+ownNodeId);
}
public void tearDown() throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/software/webapp/src/test/java/brooklyn/entity/webapp/AbstractWebAppFixtureIntegrationTest.java
----------------------------------------------------------------------
diff --git a/software/webapp/src/test/java/brooklyn/entity/webapp/AbstractWebAppFixtureIntegrationTest.java b/software/webapp/src/test/java/brooklyn/entity/webapp/AbstractWebAppFixtureIntegrationTest.java
index b8c14d7..2767444 100644
--- a/software/webapp/src/test/java/brooklyn/entity/webapp/AbstractWebAppFixtureIntegrationTest.java
+++ b/software/webapp/src/test/java/brooklyn/entity/webapp/AbstractWebAppFixtureIntegrationTest.java
@@ -59,6 +59,7 @@ import brooklyn.location.basic.LocalhostMachineProvisioningLocation;
import brooklyn.management.ManagementContext;
import brooklyn.management.SubscriptionContext;
import brooklyn.management.SubscriptionHandle;
+import brooklyn.management.ha.ManagementNodeState;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.management.internal.ManagementContextInternal;
import brooklyn.mementos.BrooklynMemento;
@@ -226,7 +227,7 @@ public abstract class AbstractWebAppFixtureIntegrationTest {
BrooklynMementoPersisterToMultiFile newPersister = new BrooklynMementoPersisterToMultiFile(tempDir , getClass().getClassLoader());
newManagementContext = new LocalManagementContextForTests();
newManagementContext.getRebindManager().setPersister(newPersister, PersistenceExceptionHandlerImpl.builder().build());
- newManagementContext.getRebindManager().rebind(getClass().getClassLoader());
+ newManagementContext.getRebindManager().rebind(getClass().getClassLoader(), null, ManagementNodeState.MASTER);
newManagementContext.getRebindManager().startPersistence();
SoftwareProcess entity2 = (SoftwareProcess) newManagementContext.getEntityManager().getEntity(tokill.getId());
entity2.stop();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/usage/launcher/src/main/java/brooklyn/launcher/BrooklynLauncher.java
----------------------------------------------------------------------
diff --git a/usage/launcher/src/main/java/brooklyn/launcher/BrooklynLauncher.java b/usage/launcher/src/main/java/brooklyn/launcher/BrooklynLauncher.java
index bea6c2e..df190d1 100644
--- a/usage/launcher/src/main/java/brooklyn/launcher/BrooklynLauncher.java
+++ b/usage/launcher/src/main/java/brooklyn/launcher/BrooklynLauncher.java
@@ -19,7 +19,6 @@
package brooklyn.launcher;
import static com.google.common.base.Preconditions.checkNotNull;
-
import brooklyn.catalog.CatalogLoadMode;
import brooklyn.internal.BrooklynFeatureEnablement;
import io.brooklyn.camp.CampPlatform;
@@ -73,6 +72,7 @@ import brooklyn.management.ManagementContext;
import brooklyn.management.ha.HighAvailabilityManager;
import brooklyn.management.ha.HighAvailabilityManagerImpl;
import brooklyn.management.ha.HighAvailabilityMode;
+import brooklyn.management.ha.ManagementNodeState;
import brooklyn.management.ha.ManagementPlaneSyncRecordPersister;
import brooklyn.management.ha.ManagementPlaneSyncRecordPersisterToObjectStore;
import brooklyn.management.internal.LocalManagementContext;
@@ -783,7 +783,7 @@ public class BrooklynLauncher {
ClassLoader classLoader = managementContext.getCatalog().getRootClassLoader();
try {
- rebindManager.rebind(classLoader);
+ rebindManager.rebind(classLoader, null, ManagementNodeState.MASTER);
} catch (Exception e) {
Exceptions.propagateIfFatal(e);
LOG.debug("Error rebinding to persisted state (rethrowing): "+e, e);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/ea4b4ccb/usage/launcher/src/test/java/brooklyn/launcher/BrooklynLauncherHighAvailabilityTest.java
----------------------------------------------------------------------
diff --git a/usage/launcher/src/test/java/brooklyn/launcher/BrooklynLauncherHighAvailabilityTest.java b/usage/launcher/src/test/java/brooklyn/launcher/BrooklynLauncherHighAvailabilityTest.java
index 419124d..f151d92 100644
--- a/usage/launcher/src/test/java/brooklyn/launcher/BrooklynLauncherHighAvailabilityTest.java
+++ b/usage/launcher/src/test/java/brooklyn/launcher/BrooklynLauncherHighAvailabilityTest.java
@@ -130,7 +130,8 @@ public class BrooklynLauncherHighAvailabilityTest {
ManagementContext secondaryManagementContext = secondary.getServerDetails().getManagementContext();
log.info("started mgmt secondary "+secondaryManagementContext);
- assertNoApps(secondary.getServerDetails().getManagementContext());
+ // TODO can assert it sees the apps read only
+// assertNoApps(secondary.getServerDetails().getManagementContext());
// Terminate primary; expect secondary to take over
if (stopGracefully) {
@@ -143,11 +144,11 @@ public class BrooklynLauncherHighAvailabilityTest {
assertOnlyAppEventually(secondaryManagementContext, TestApplication.class);
- // Start tertiary (will come up as standby)
+ // Start tertiary (force up as standby)
tertiary = BrooklynLauncher.newInstance();
tertiary.webconsole(false)
.brooklynProperties(LocalManagementContextForTests.setEmptyCatalogAsDefault(BrooklynProperties.Factory.newEmpty()))
- .highAvailabilityMode(HighAvailabilityMode.AUTO)
+ .highAvailabilityMode(HighAvailabilityMode.STANDBY)
.persistMode(PersistMode.AUTO)
.persistenceDir(persistenceDir)
.persistPeriod(Duration.millis(10))