You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by ri...@apache.org on 2014/06/16 21:45:20 UTC
[3/5] git commit: change so persistence dir is only backed up if node
is HA master or HA disabled;
handles the case of HA autodetect selecting master hence the complexity. this
is done by introducing a "deferredBackupNeeded" flag in the ObjectStore which
change so persistence dir is only backed up if node is HA master or HA disabled;
handles the case of HA autodetect selecting master hence the complexity.
this is done by introducing a "deferredBackupNeeded" flag in the ObjectStore which kicks in when triggered by certain "contended writes",
e.g. just after a standby node is promoted, or auto detects master. (Previously tried to detect if master/standby then backup immediately but
was too racy to try to let persister read if-master from the store while the store is waiting to know if-master!)
also:
* fixes some leaks in test
* improves atomic file update support in FileBasedObjectStore.
* adds startup flags to allow ignoring various types of errors, with default being to stop on any of the errors
(the one irritating corner case is if persistence is on and we start an app which fails,
those failed apps remain persisted and you see the failures when you successfully start ...
that's probably the right behaviour but will be irritating for some people;
might in future want a --destroyFailedStartupApps CLI option)
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/3c444b47
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/3c444b47
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/3c444b47
Branch: refs/heads/master
Commit: 3c444b4716ced93ac4c0ed0a5fecb3edc216da97
Parents: 6de7537
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Sat Jun 14 19:28:35 2014 -0700
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Sat Jun 14 20:14:07 2014 -0700
----------------------------------------------------------------------
.../management/ha/HighAvailabilityManager.java | 7 +-
.../entity/basic/BrooklynShutdownHooks.java | 201 ++++++++++++++-----
.../java/brooklyn/entity/basic/Entities.java | 4 +-
.../brooklyn/entity/basic/EntityFunctions.java | 1 +
.../BrooklynMementoPersisterToObjectStore.java | 2 +
.../rebind/persister/FileBasedObjectStore.java | 131 ++++++++++--
.../persister/PersistenceObjectStore.java | 25 ++-
.../persister/StoreObjectAccessorLocking.java | 7 +-
.../location/geo/LocalhostExternalIpLoader.java | 41 ++--
.../location/geo/MaxMindHostGeoLookup.java | 2 +-
.../location/geo/UtraceHostGeoLookup.java | 14 +-
.../ha/HighAvailabilityManagerImpl.java | 14 +-
...ntPlaneSyncRecordPersisterToObjectStore.java | 56 ++++--
.../internal/LocalManagementContext.java | 2 +-
.../NonDeploymentManagementContext.java | 2 +
.../brooklyn/util/BrooklynNetworkUtils.java | 6 +-
.../entity/basic/BrooklynShutdownHooksTest.java | 9 +-
.../entity/group/DynamicMultiGroupTest.java | 2 +-
.../entity/rebind/RebindEntityTest.java | 4 +-
.../entity/rebind/RebindTestFixture.java | 10 +-
.../brooklyn/entity/rebind/RebindTestUtils.java | 6 +-
.../rebind/persister/InMemoryObjectStore.java | 16 +-
.../InMemoryStoreObjectAccessorWriterTest.java | 4 +-
.../rebind/persister/ListeningObjectStore.java | 22 +-
...ocalhostExternalIpLoaderIntegrationTest.java | 2 +-
.../ha/HighAvailabilityManagerTestFixture.java | 3 +-
.../JcloudsBlobStoreBasedObjectStore.java | 33 ++-
.../JcloudsLocationSecurityGroupCustomizer.java | 5 +-
.../JcloudsObjectStoreAccessorWriterTest.java | 4 +-
usage/cli/src/main/java/brooklyn/cli/Main.java | 93 +++++++--
.../brooklyn/launcher/BrooklynLauncher.java | 162 ++++++++-------
.../brooklyn/launcher/BrooklynWebServer.java | 2 +-
.../config/StopWhichAppsOnShutdown.java | 5 +
.../BrooklynLauncherHighAvailabilityTest.java | 58 ++++--
.../src/main/java/brooklyn/util/ShellUtils.java | 3 +-
.../brooklyn/util/exceptions/Exceptions.java | 15 +-
36 files changed, 694 insertions(+), 279 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/api/src/main/java/brooklyn/management/ha/HighAvailabilityManager.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/brooklyn/management/ha/HighAvailabilityManager.java b/api/src/main/java/brooklyn/management/ha/HighAvailabilityManager.java
index 46e5a1b..06536f1 100644
--- a/api/src/main/java/brooklyn/management/ha/HighAvailabilityManager.java
+++ b/api/src/main/java/brooklyn/management/ha/HighAvailabilityManager.java
@@ -51,9 +51,10 @@ public interface HighAvailabilityManager {
/**
* Starts the monitoring of other nodes (and thus potential promotion of this node from standby to master).
* <p>
- * By the time this method returns, then if appropriate this node will already be {@link ManagementNodeState#MASTER}.
- * Otherwise it will be {@link ManagementNodeState#STANDBY}.
- *
+ * When this method returns, the status of this node will be set,
+ * either {@link ManagementNodeState#MASTER} if appropriate or {@link ManagementNodeState#STANDBY}.
+ *
+ * @param startMode mode to start with
* @throws IllegalStateException if current state of the management-plane doesn't match that desired by {@code startMode}
*/
void start(HighAvailabilityMode startMode);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/entity/basic/BrooklynShutdownHooks.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/basic/BrooklynShutdownHooks.java b/core/src/main/java/brooklyn/entity/basic/BrooklynShutdownHooks.java
index d6fe3b3..44882ec 100644
--- a/core/src/main/java/brooklyn/entity/basic/BrooklynShutdownHooks.java
+++ b/core/src/main/java/brooklyn/entity/basic/BrooklynShutdownHooks.java
@@ -1,7 +1,8 @@
package brooklyn.entity.basic;
-import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
@@ -12,10 +13,13 @@ import brooklyn.entity.trait.Startable;
import brooklyn.management.ManagementContext;
import brooklyn.management.Task;
import brooklyn.management.internal.ManagementContextInternal;
-import brooklyn.util.collections.MutableMap;
+import brooklyn.util.collections.MutableList;
+import brooklyn.util.collections.MutableSet;
import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.exceptions.RuntimeInterruptedException;
import brooklyn.util.javalang.Threads;
+import brooklyn.util.task.Tasks;
+import brooklyn.util.time.CountdownTimer;
import brooklyn.util.time.Duration;
import com.google.common.annotations.VisibleForTesting;
@@ -29,9 +33,12 @@ public class BrooklynShutdownHooks {
private static final AtomicBoolean isShutdownHookRegistered = new AtomicBoolean();
private static final List<Entity> entitiesToStopOnShutdown = Lists.newArrayList();
- private static final List<ManagementContextInternal> managementContextsToTerminateOnShutdown = Lists.newArrayList();
+ private static final List<ManagementContext> managementContextsToStopAppsOnShutdown = Lists.newArrayList();
+ private static final List<ManagementContext> managementContextsToTerminateOnShutdown = Lists.newArrayList();
+ private static final AtomicBoolean isShutDown = new AtomicBoolean(false);
- private static final Object mutex = new Object();
+// private static final Object mutex = new Object();
+ private static final Semaphore semaphore = new Semaphore(1);
/**
* Max time to wait for shutdown to complete, when stopping the entities from {@link #invokeStopOnShutdown(Entity)}.
@@ -48,70 +55,168 @@ public class BrooklynShutdownHooks {
log.warn("Not adding entity {} for stop-on-shutdown as not an instance of {}", entity, Startable.class.getSimpleName());
return;
}
- synchronized (mutex) {
+ try {
+ semaphore.acquire();
+ if (isShutDown.get()) {
+ semaphore.release();
+ try {
+ log.warn("Call to invokeStopOnShutdown for "+entity+" while system already shutting down; invoking stop now and throwing exception");
+ Entities.destroy(entity);
+ throw new IllegalStateException("Call to invokeStopOnShutdown for "+entity+" while system already shutting down");
+ } catch (Exception e) {
+ throw new IllegalStateException("Call to invokeStopOnShutdown for "+entity+" while system already shutting down, had error: "+e, e);
+ }
+ }
+
+ // TODO should be a weak reference in case it is destroyed before shutdown
+ // (only applied to certain entities started via launcher so not a big leak)
entitiesToStopOnShutdown.add(entity);
+ semaphore.release();
+ addShutdownHookIfNotAlready();
+
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
}
- addShutdownHookIfNotAlready();
}
-
+
+ public static void invokeStopAppsOnShutdown(ManagementContext managementContext) {
+ try {
+ semaphore.acquire();
+ if (isShutDown.get()) {
+ semaphore.release();
+ try {
+ log.warn("Call to invokeStopAppsOnShutdown for "+managementContext+" while system already shutting down; invoking stop now and throwing exception");
+ destroyAndWait(managementContext.getApplications(), shutdownTimeout);
+
+ throw new IllegalStateException("Call to invokeStopAppsOnShutdown for "+managementContext+" while system already shutting down");
+ } catch (Exception e) {
+ throw new IllegalStateException("Call to invokeStopAppsOnShutdown for "+managementContext+" while system already shutting down, had error: "+e, e);
+ }
+ }
+
+ // TODO weak reference, as per above
+ managementContextsToStopAppsOnShutdown.add(managementContext);
+ semaphore.release();
+ addShutdownHookIfNotAlready();
+
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ }
+
public static void invokeTerminateOnShutdown(ManagementContext managementContext) {
- synchronized (mutex) {
- managementContextsToTerminateOnShutdown.add((ManagementContextInternal) managementContext);
+ try {
+ semaphore.acquire();
+ if (isShutDown.get()) {
+ semaphore.release();
+ try {
+ log.warn("Call to invokeStopOnShutdown for "+managementContext+" while system already shutting down; invoking stop now and throwing exception");
+ ((ManagementContextInternal)managementContext).terminate();
+ throw new IllegalStateException("Call to invokeTerminateOnShutdown for "+managementContext+" while system already shutting down");
+ } catch (Exception e) {
+ throw new IllegalStateException("Call to invokeTerminateOnShutdown for "+managementContext+" while system already shutting down, had error: "+e, e);
+ }
+ }
+
+ // TODO weak reference, as per above
+ managementContextsToTerminateOnShutdown.add(managementContext);
+ semaphore.release();
+ addShutdownHookIfNotAlready();
+
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
}
- addShutdownHookIfNotAlready();
}
private static void addShutdownHookIfNotAlready() {
if (isShutdownHookRegistered.compareAndSet(false, true)) {
- Threads.addShutdownHook(new BrooklynShutdownHookJob());
+ Threads.addShutdownHook(BrooklynShutdownHookJob.newInstanceForReal());
}
}
@VisibleForTesting
public static class BrooklynShutdownHookJob implements Runnable {
- @SuppressWarnings({ "unchecked", "rawtypes" })
+
+ final boolean setStaticShutDownFlag;
+
+ private BrooklynShutdownHookJob(boolean setStaticShutDownFlag) {
+ this.setStaticShutDownFlag = setStaticShutDownFlag;
+ }
+
+ public static BrooklynShutdownHookJob newInstanceForReal() {
+ return new BrooklynShutdownHookJob(true);
+ }
+
+ /** testing instance does not actually set the `isShutDown` bit */
+ public static BrooklynShutdownHookJob newInstanceForTesting() {
+ return new BrooklynShutdownHookJob(false);
+ }
+
@Override
public void run() {
// First stop entities; on interrupt, abort waiting for tasks - but let shutdown hook continue
- synchronized (mutex) {
- if (entitiesToStopOnShutdown.isEmpty())
- log.debug("Brooklyn stopOnShutdown shutdown-hook invoked: no entities to stop");
- else
- log.info("Brooklyn stopOnShutdown shutdown-hook invoked: stopping entities: "+entitiesToStopOnShutdown);
- List<Task> stops = new ArrayList<Task>();
- for (Entity entity: entitiesToStopOnShutdown) {
- try {
- stops.add(entity.invoke(Startable.STOP, new MutableMap()));
- } catch (RuntimeException exc) {
- if (log.isDebugEnabled()) log.debug("stopOnShutdown of "+entity+" returned error (continuing): "+exc, exc);
- }
- }
- long endTime = System.currentTimeMillis() + shutdownTimeout.toMilliseconds();
- for (Task t: stops) {
- Duration remainingTime = Duration.max(Duration.untilUtc(endTime), Duration.ONE_MILLISECOND);
- try {
- Object result = t.getUnchecked(remainingTime);
- if (log.isDebugEnabled()) log.debug("stopOnShutdown of {} completed: {}", t, result);
- } catch (RuntimeInterruptedException e) {
- Thread.currentThread().interrupt();
- endTime = System.currentTimeMillis();
- if (log.isDebugEnabled()) log.debug("stopOnShutdown of "+t+" interrupted; (continuing with immediate timeout): "+e);
- } catch (RuntimeException e) {
- if (log.isDebugEnabled()) log.debug("stopOnShutdown of "+t+" returned error (continuing): "+e, e);
- Exceptions.propagateIfFatal(e);
- }
- }
+ Set<Entity> entitiesToStop = MutableSet.of();
+ try {
+ semaphore.acquire();
+ if (setStaticShutDownFlag)
+ isShutDown.set(true);
+ semaphore.release();
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ for (Entity entity: entitiesToStopOnShutdown)
+ entitiesToStop.add(entity);
+ for (ManagementContext mgmt: managementContextsToStopAppsOnShutdown) {
+ entitiesToStop.addAll(mgmt.getApplications());
+ }
- // Then terminate management contexts
- log.debug("Brooklyn terminateOnShutdown shutdown-hook invoked: terminating management contexts: "+managementContextsToTerminateOnShutdown);
- for (ManagementContextInternal managementContext: managementContextsToTerminateOnShutdown) {
- try {
- managementContext.terminate();
- } catch (RuntimeException e) {
- log.info("terminateOnShutdown of "+managementContext+" returned error (continuing): "+e, e);
- }
+ if (entitiesToStop.isEmpty()) {
+ log.debug("Brooklyn shutdown: no entities to stop");
+ } else {
+ log.info("Brooklyn stopdown: stopping entities "+entitiesToStop);
+ destroyAndWait(entitiesToStop, shutdownTimeout);
+ }
+
+ // Then terminate management contexts
+ log.debug("Brooklyn terminateOnShutdown shutdown-hook invoked: terminating management contexts: "+managementContextsToTerminateOnShutdown);
+ for (ManagementContext managementContext: managementContextsToTerminateOnShutdown) {
+ try {
+ if (!managementContext.isRunning())
+ continue;
+ ((ManagementContextInternal)managementContext).terminate();
+ } catch (RuntimeException e) {
+ log.info("terminateOnShutdown of "+managementContext+" returned error (continuing): "+e, e);
}
}
}
}
+
+ protected static void destroyAndWait(Iterable<? extends Entity> entitiesToStop, Duration timeout) {
+ MutableList<Task<?>> stops = MutableList.of();
+ for (Entity entityToStop: entitiesToStop) {
+ final Entity entity = entityToStop;
+ if (!Entities.isManaged(entity)) continue;
+ Task<Object> t = Tasks.builder().dynamic(false).name("destroying "+entity).body(new Runnable() {
+ @Override public void run() { Entities.destroy(entity); }
+ }).build();
+ stops.add( ((EntityInternal)entity).getExecutionContext().submit(t) );
+ }
+ CountdownTimer timer = CountdownTimer.newInstanceStarted(timeout);
+ for (Task<?> t: stops) {
+ try {
+ Duration durationRemaining = timer.getDurationRemaining();
+ Object result = t.getUnchecked(durationRemaining.isPositive() ? durationRemaining : Duration.ONE_MILLISECOND);
+ if (log.isDebugEnabled()) log.debug("stopOnShutdown of {} completed: {}", t, result);
+ } catch (RuntimeInterruptedException e) {
+ Thread.currentThread().interrupt();
+ if (log.isDebugEnabled()) log.debug("stopOnShutdown of "+t+" interrupted: "+e);
+ break;
+ } catch (RuntimeException e) {
+ Exceptions.propagateIfFatal(e);
+ log.warn("Shutdown hook "+t+" returned error (continuing): "+e);
+ if (log.isDebugEnabled()) log.debug("stopOnShutdown of "+t+" returned error (continuing to stop others): "+e, e);
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 b27fefe..48ce3ed 100644
--- a/core/src/main/java/brooklyn/entity/basic/Entities.java
+++ b/core/src/main/java/brooklyn/entity/basic/Entities.java
@@ -510,9 +510,9 @@ public class Entities {
}
/**
- * @deprecated since 0.7.0; instead use {@link BrooklynShutdownHooks#invokeStopOnShutdown(Entity)}
+ * Registers a {@link BrooklynShutdownHooks#invokeStopOnShutdown(Entity)} to shutdown this entity when the JVM exits.
+ * (Convenience method located in this class for easy access.)
*/
- @Deprecated
public static void invokeStopOnShutdown(Entity entity) {
BrooklynShutdownHooks.invokeStopOnShutdown(entity);
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/entity/basic/EntityFunctions.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/basic/EntityFunctions.java b/core/src/main/java/brooklyn/entity/basic/EntityFunctions.java
index 0be694d..725bf7f 100644
--- a/core/src/main/java/brooklyn/entity/basic/EntityFunctions.java
+++ b/core/src/main/java/brooklyn/entity/basic/EntityFunctions.java
@@ -13,6 +13,7 @@ import brooklyn.management.ManagementContext;
import brooklyn.util.flags.TypeCoercions;
import com.google.common.base.Function;
+import com.google.common.base.Predicate;
import com.google.common.base.Supplier;
public class EntityFunctions {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 a884419..820f54d 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/BrooklynMementoPersisterToObjectStore.java
@@ -267,6 +267,7 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
return;
}
if (LOG.isDebugEnabled()) LOG.debug("Checkpointing entire memento");
+ objectStore.prepareForContendedWrite();
for (EntityMemento entity : newMemento.getEntityMementos().values()) {
persist("entities", entity);
@@ -292,6 +293,7 @@ public class BrooklynMementoPersisterToObjectStore implements BrooklynMementoPer
"removing {} entities, {} locations and {} policies",
new Object[] {delta.entities(), delta.locations(), delta.policies(),
delta.removedEntityIds(), delta.removedLocationIds(), delta.removedPolicyIds()});
+ objectStore.prepareForContendedWrite();
for (EntityMemento entity : delta.entities()) {
persist("entities", entity);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java b/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
index e9a12b5..7e4bd82 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/FileBasedObjectStore.java
@@ -11,6 +11,7 @@ import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nullable;
@@ -20,14 +21,20 @@ import org.slf4j.LoggerFactory;
import brooklyn.config.BrooklynServerConfig;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
+import brooklyn.util.collections.MutableList;
+import brooklyn.util.collections.MutableMap;
import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.exceptions.FatalConfigurationRuntimeException;
+import brooklyn.util.internal.ssh.process.ProcessTool;
import brooklyn.util.os.Os;
import brooklyn.util.os.Os.DeletionResult;
import com.google.common.base.Function;
import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
import com.google.common.collect.FluentIterable;
+import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
@@ -41,10 +48,11 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
private static final int SHUTDOWN_TIMEOUT_MS = 10*1000;
private final File basedir;
-
private final ListeningExecutorService executor;
-
private ManagementContext mgmt;
+ private boolean prepared = false;
+ private boolean deferredBackupNeeded = false;
+ private AtomicBoolean doneFirstContentiousWrite = new AtomicBoolean(false);
/**
* @param basedir
@@ -65,15 +73,44 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
return basedir;
}
+ public void prepareForContendedWrite() {
+ if (doneFirstContentiousWrite.get())
+ return;
+ synchronized (this) {
+ if (doneFirstContentiousWrite.get())
+ return;
+ try {
+ if (deferredBackupNeeded) {
+ // defer backup and path creation until first write
+ // this way if node is standby or auto, the backup is not created superfluously
+
+ File backup = backupDirByCopying(basedir);
+ log.info("Persistence deferred backup, directory "+basedir+" backed up to "+backup.getAbsolutePath());
+
+ deferredBackupNeeded = false;
+ }
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ doneFirstContentiousWrite.getAndSet(true);
+ }
+ }
+
@Override
public void createSubPath(String subPath) {
+ if (!prepared) throw new IllegalStateException("Not yet prepared: "+this);
+
File dir = new File(getBaseDir(), subPath);
- dir.mkdir();
+ if (!dir.mkdir())
+ if (!dir.exists())
+ throw new IllegalStateException("Cannot create "+dir+"; call returned false");
checkPersistenceDirAccessible(dir);
}
@Override
public StoreObjectAccessor newAccessor(String path) {
+ if (!prepared) throw new IllegalStateException("Not yet prepared: "+this);
+
String tmpExt = ".tmp";
if (mgmt!=null && mgmt.getManagementNodeId()!=null) tmpExt = "."+mgmt.getManagementNodeId()+tmpExt;
return new FileBasedStoreObjectAccessor(new File(Os.mergePaths(getBaseDir().getAbsolutePath(), path)), tmpExt);
@@ -81,6 +118,9 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
@Override
public List<String> listContentsWithSubPath(final String parentSubPath) {
+ if (!prepared) throw new IllegalStateException("Not yet prepared: "+this);
+
+ Preconditions.checkNotNull(parentSubPath);
File subPathDir = new File(basedir, parentSubPath);
FileFilter fileFilter = new FileFilter() {
@@ -89,6 +129,7 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
}
};
File[] subPathDirFiles = subPathDir.listFiles(fileFilter);
+ if (subPathDirFiles==null) return ImmutableList.<String>of();
return FluentIterable.from(Arrays.asList(subPathDirFiles))
.transform(new Function<File, String>() {
@Nullable
@@ -115,14 +156,21 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
}
@Override
- public void prepareForUse(ManagementContext mgmt, @Nullable PersistMode persistMode) {
+ public void injectManagementContext(ManagementContext mgmt) {
if (this.mgmt!=null && !this.mgmt.equals(mgmt))
throw new IllegalStateException("Cannot change mgmt context of "+this);
this.mgmt = mgmt;
-
- if (persistMode==null || persistMode==PersistMode.DISABLED)
+ }
+
+ @Override
+ public void prepareForUse(@Nullable PersistMode persistMode, HighAvailabilityMode haMode) {
+ if (mgmt==null) throw new NullPointerException("Must inject ManagementContext before preparing "+this);
+
+ if (persistMode==null || persistMode==PersistMode.DISABLED) {
// is this check needed? shouldn't come here now without persistence on.
+ prepared = true;
return;
+ }
Boolean backups = mgmt.getConfig().getConfig(BrooklynServerConfig.PERSISTENCE_BACKUPS_REQUIRED);
if (backups==null) backups = true; // for file system
@@ -155,8 +203,12 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
checkPersistenceDirNonEmpty(dir);
try {
if (backups) {
- File backup = backupDirByCopying(dir);
- log.info("Persistence mode REBIND, directory "+persistencePath+" backed up to "+backup.getAbsolutePath());
+ if (haMode==HighAvailabilityMode.MASTER) {
+ File backup = backupDirByCopying(dir);
+ log.info("Persistence mode REBIND, directory "+persistencePath+" backed up to "+backup.getAbsolutePath());
+ } else {
+ deferredBackupNeeded = true;
+ }
}
} catch (IOException e) {
throw new FatalConfigurationRuntimeException("Error backing up persistence directory "+dir.getAbsolutePath(), e);
@@ -169,8 +221,12 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
if (dir.exists() && !isMementoDirExistButEmpty(dir)) {
try {
if (backups) {
- File backup = backupDirByCopying(dir);
- log.info("Persistence mode REBIND, directory "+persistencePath+" backed up to "+backup.getAbsolutePath());
+ if (haMode==HighAvailabilityMode.MASTER) {
+ File backup = backupDirByCopying(dir);
+ log.info("Persistence mode REBIND, directory "+persistencePath+" backed up to "+backup.getAbsolutePath());
+ } else {
+ deferredBackupNeeded = true;
+ }
}
} catch (IOException e) {
throw new FatalConfigurationRuntimeException("Error backing up persistence directory "+dir.getAbsolutePath(), e);
@@ -193,32 +249,40 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
} catch (Exception e) {
throw Exceptions.propagate(e);
}
+
+ prepared = true;
}
protected void checkPersistenceDirAccessible(File dir) {
if (!(dir.exists() && dir.isDirectory() && dir.canRead() && dir.canWrite())) {
- throw new FatalConfigurationRuntimeException("Invalid persistence directory " + dir + ": " +
+ FatalConfigurationRuntimeException problem = new FatalConfigurationRuntimeException("Invalid persistence directory " + dir + ": " +
(!dir.exists() ? "does not exist" :
(!dir.isDirectory() ? "not a directory" :
(!dir.canRead() ? "not readable" :
(!dir.canWrite() ? "not writable" : "unknown reason")))));
+ log.debug("Invalid persistence directory "+dir+" (rethrowing): "+problem, problem);
} else {
log.debug("Created dir {} for {}", dir, this);
}
}
protected void checkPersistenceDirNonEmpty(File persistenceDir) {
- if (!persistenceDir.exists())
- throw new FatalConfigurationRuntimeException("Invalid persistence directory "+persistenceDir+" because directory does not exist");
- if (isMementoDirExistButEmpty(persistenceDir)) {
- throw new FatalConfigurationRuntimeException("Invalid persistence directory "+persistenceDir+" because directory is empty");
+ FatalConfigurationRuntimeException problem;
+ if (!persistenceDir.exists()) {
+ problem = new FatalConfigurationRuntimeException("Invalid persistence directory "+persistenceDir+" because directory does not exist");
+ log.debug("Invalid persistence directory "+persistenceDir+" (rethrowing): "+problem, problem);
+ throw problem;
+ } if (isMementoDirExistButEmpty(persistenceDir)) {
+ problem = new FatalConfigurationRuntimeException("Invalid persistence directory "+persistenceDir+" because directory is empty");
+ log.debug("Invalid persistence directory "+persistenceDir+" (rethrowing): "+problem, problem);
+ throw problem;
}
}
protected File backupDirByCopying(File dir) throws IOException, InterruptedException {
File parentDir = dir.getParentFile();
String simpleName = dir.getName();
- String timestamp = new SimpleDateFormat("yyyy-MM-dd-hhmm-ss").format(new Date());
+ String timestamp = new SimpleDateFormat("yyyyMMdd-hhmmssSSS").format(new Date());
File backupDir = new File(parentDir, simpleName+"-"+timestamp+".bak");
copyDir(dir, backupDir);
@@ -228,13 +292,14 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
protected File backupDirByMoving(File dir) throws InterruptedException, IOException {
File parentDir = dir.getParentFile();
String simpleName = dir.getName();
- String timestamp = new SimpleDateFormat("yyyy-MM-dd-hhmm-ss").format(new Date());
+ String timestamp = new SimpleDateFormat("yyyyMMdd-hhmmssSSS").format(new Date());
File newDir = new File(parentDir, simpleName+"-"+timestamp+".old");
moveDir(dir, newDir);
return newDir;
}
+ private static boolean WARNED_ON_NON_ATOMIC_FILE_UPDATES = false;
/**
* Attempts an fs level atomic move then fall back to pure java rename.
* Assumes files are on same mount point.
@@ -243,14 +308,38 @@ public class FileBasedObjectStore implements PersistenceObjectStore {
*/
static void moveFile(File srcFile, File destFile) throws IOException, InterruptedException {
if (!Os.isMicrosoftWindows()) {
+ // this command, if it succeeds, is guaranteed to be atomic, and it will usually overwrite
+ int result;
String cmd = "mv '"+srcFile.getAbsolutePath()+"' '"+destFile.getAbsolutePath()+"'";
- Process proc = Runtime.getRuntime().exec(cmd);
- proc.waitFor();
- if (proc.exitValue() == 0) return;
+
+ result = new ProcessTool().execCommands(MutableMap.<String,String>of(), MutableList.of(cmd), null);
+ // prefer the above to the below because it wraps it in the appropriate bash
+// Process proc = Runtime.getRuntime().exec(cmd);
+// result = proc.waitFor();
+
+ log.trace("FS move of {} to {} completed, code {}", new Object[] { srcFile, destFile, result });
+ if (result == 0) return;
}
+ boolean result;
+ // this is not guaranteed cross platform to succeed if the destination exists,
+ // and not guaranteed to be atomic, but it usually seems to do the right thing...
+ result = srcFile.renameTo(destFile);
+ if (result) {
+ log.trace("java rename of {} to {} completed", srcFile, destFile);
+ return;
+ }
+
+ // finally try a delete - but explicitly warn this is not going to be atomic
+ // so if another node reads it might see no master
+ if (!WARNED_ON_NON_ATOMIC_FILE_UPDATES) {
+ WARNED_ON_NON_ATOMIC_FILE_UPDATES = true;
+ log.warn("Unable to perform atomic file update ("+srcFile+" to "+destFile+"); file system not recommended for production HA/DR");
+ }
destFile.delete();
- srcFile.renameTo(destFile);
+ result = srcFile.renameTo(destFile);
+ log.trace("java delete and rename of {} to {} completed, code {}", new Object[] { srcFile, destFile, result });
+ if (!result) throw new IOException("Could not rename "+destFile+" to "+srcFile);
}
static void moveDir(File srcDir, File destDir) throws IOException, InterruptedException {
if (!Os.isMicrosoftWindows()) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/entity/rebind/persister/PersistenceObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/PersistenceObjectStore.java b/core/src/main/java/brooklyn/entity/rebind/persister/PersistenceObjectStore.java
index c75c6a1..1e2bd9d 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/PersistenceObjectStore.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/PersistenceObjectStore.java
@@ -5,6 +5,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReadWriteLock;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.util.time.Duration;
import com.google.common.annotations.Beta;
@@ -40,6 +41,11 @@ public interface PersistenceObjectStore {
/** human-readable name of this object store */
public String getSummaryName();
+ /** triggers any backup if this method has not been invoked;
+ * used to allow certain non-contentious writes (e.g. node id's)
+ * without triggering backup, but then forcing backup before a contended write */
+ public void prepareForContendedWrite();
+
/**
* For reading/writing data to the item at the given path.
* Note that the accessor is not generally thread safe, usually does not support blocking,
@@ -75,18 +81,21 @@ public interface PersistenceObjectStore {
void close();
/**
- * Prepares the persistence directory for use (e.g. backing up old dir, checking is non-empty
- * or deleting as required, creating, etc).
- * <p>
- * Also allows a way for an object store to be created ahead of time,
- * and a mgmt context injected.
- * (If persistMode is null, it is ignored.)
+ * Allows a way for an object store to be created ahead of time, and a mgmt context injected.
+ * Currently subsequent changes are not permitted.
* <p>
- * currently subsequent changes are not permitted.
+ * A {@link ManagementContext} must be supplied via constructor or this method before invoking other methods.
+ */
+ @Beta
+ public void injectManagementContext(ManagementContext managementContext);
+
+ /**
+ * Prepares the persistence directory for use
+ * (e.g. detecting whether any backup will be necessary, deleting as required, etc)
*/
// although there is some commonality between the different stores it is mostly different,
// so this method currently sits here; it may move if advanced backup strategies have commonalities
@Beta
- public void prepareForUse(ManagementContext managementContext, PersistMode persistMode);
+ public void prepareForUse(PersistMode persistMode, HighAvailabilityMode highAvailabilityMode);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/entity/rebind/persister/StoreObjectAccessorLocking.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/rebind/persister/StoreObjectAccessorLocking.java b/core/src/main/java/brooklyn/entity/rebind/persister/StoreObjectAccessorLocking.java
index 1e94703..a3688bb 100644
--- a/core/src/main/java/brooklyn/entity/rebind/persister/StoreObjectAccessorLocking.java
+++ b/core/src/main/java/brooklyn/entity/rebind/persister/StoreObjectAccessorLocking.java
@@ -8,6 +8,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.javalang.JavaClassNames;
import brooklyn.util.time.Duration;
public class StoreObjectAccessorLocking implements PersistenceObjectStore.StoreObjectAccessorWithLock {
@@ -149,5 +150,9 @@ public class StoreObjectAccessorLocking implements PersistenceObjectStore.StoreO
public ReadWriteLock getLockObject() {
return lock;
}
-
+
+ @Override
+ public String toString() {
+ return JavaClassNames.simpleClassName(this)+":"+delegate.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/location/geo/LocalhostExternalIpLoader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/location/geo/LocalhostExternalIpLoader.java b/core/src/main/java/brooklyn/location/geo/LocalhostExternalIpLoader.java
index fb7199f..11bae12 100644
--- a/core/src/main/java/brooklyn/location/geo/LocalhostExternalIpLoader.java
+++ b/core/src/main/java/brooklyn/location/geo/LocalhostExternalIpLoader.java
@@ -1,11 +1,9 @@
package brooklyn.location.geo;
-import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -13,15 +11,14 @@ import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
-import com.google.common.base.Splitter;
-import com.google.common.base.Throwables;
-import com.google.common.collect.Lists;
-
import brooklyn.util.ResourceUtils;
import brooklyn.util.exceptions.Exceptions;
import brooklyn.util.time.Duration;
+import brooklyn.util.time.Durations;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
public class LocalhostExternalIpLoader {
@@ -75,10 +72,17 @@ public class LocalhostExternalIpLoader {
static String getIpAddressFrom(String url) {
return new IpLoader(url).call();
}
+
+ /** As {@link #getLocalhostIpWithin(Duration)} but returning 127.0.0.1 if not accessible */
+ public static String getLocalhostIpQuicklyOrDefault() {
+ String result = doLoad(Duration.seconds(2));
+ if (result==null) return "127.0.0.1";
+ return result;
+ }
- /** As {@link #getLocalhostIpWithin(Duration)} but without the time limit cut-off. */
- public static String getLocalhostIp() {
- return doLoad(Optional.<Duration>absent());
+ /** As {@link #getLocalhostIpWithin(Duration)} but without the time limit cut-off, failing if the load gives an error. */
+ public static String getLocalhostIpWaiting() {
+ return getLocalhostIpWithin(null);
}
/**
@@ -87,17 +91,19 @@ public class LocalhostExternalIpLoader {
* @return The public IP address of localhost
*/
public static String getLocalhostIpWithin(Duration timeout) {
- return doLoad(Optional.of(timeout));
+ String result = doLoad(timeout);
+ if (result!=null) return null;
+ throw new IllegalStateException("Unable to retrieve external IP for localhost; network may be down or slow or remote service otherwise not responding");
}
/**
* Requests URLs returned by {@link #getIpAddressWebsites()} until one returns an IP address.
* The address is assumed to be the external IP address of localhost.
* @param blockFor The maximum duration to wait for the IP address to be resolved.
- * An indefinite way if absent.
+ * An indefinite way if null.
* @return A string in IPv4 format, or null if no such address could be ascertained.
*/
- private static String doLoad(Optional<Duration> blockFor) {
+ private static String doLoad(Duration blockFor) {
if (localExternalIp != null) {
return localExternalIp;
}
@@ -131,9 +137,8 @@ public class LocalhostExternalIpLoader {
}
try {
- if (blockFor.isPresent()) {
- long millis = blockFor.get().toMilliseconds();
- triedLocalExternalIp.await(millis, TimeUnit.MILLISECONDS);
+ if (blockFor!=null) {
+ Durations.await(triedLocalExternalIp, blockFor);
} else {
triedLocalExternalIp.await();
}
@@ -141,7 +146,7 @@ public class LocalhostExternalIpLoader {
throw Exceptions.propagate(e);
}
if (localExternalIp == null) {
- throw Throwables.propagate(new IOException("Unable to discover external IP of local machine; response to server timed out (ongoing=" + retrievingLocalExternalIp + ")"));
+ return null;
}
LOG.debug("Looked up external IP of this host, result is: {}", localExternalIp);
return localExternalIp;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/location/geo/MaxMindHostGeoLookup.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/location/geo/MaxMindHostGeoLookup.java b/core/src/main/java/brooklyn/location/geo/MaxMindHostGeoLookup.java
index 2c02a90..9152a0b 100644
--- a/core/src/main/java/brooklyn/location/geo/MaxMindHostGeoLookup.java
+++ b/core/src/main/java/brooklyn/location/geo/MaxMindHostGeoLookup.java
@@ -42,7 +42,7 @@ public class MaxMindHostGeoLookup implements HostGeoLookup {
if (ll==null) return null;
InetAddress extAddress = address;
- if (Networking.isPrivateSubnet(extAddress)) extAddress = InetAddress.getByName(UtraceHostGeoLookup.getLocalhostExternalIp());
+ if (Networking.isPrivateSubnet(extAddress)) extAddress = InetAddress.getByName(LocalhostExternalIpLoader.getLocalhostIpQuicklyOrDefault());
com.maxmind.geoip.Location l = ll.getLocation(extAddress);
if (l==null) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/location/geo/UtraceHostGeoLookup.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/location/geo/UtraceHostGeoLookup.java b/core/src/main/java/brooklyn/location/geo/UtraceHostGeoLookup.java
index 09292ad..4b1c9c3 100644
--- a/core/src/main/java/brooklyn/location/geo/UtraceHostGeoLookup.java
+++ b/core/src/main/java/brooklyn/location/geo/UtraceHostGeoLookup.java
@@ -1,6 +1,5 @@
package brooklyn.location.geo;
-import brooklyn.util.time.Duration;
import groovy.util.Node;
import groovy.util.NodeList;
import groovy.util.XmlParser;
@@ -8,17 +7,12 @@ import groovy.util.XmlParser;
import java.io.IOException;
import java.net.InetAddress;
import java.net.MalformedURLException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import brooklyn.util.exceptions.Exceptions;
-import brooklyn.util.ResourceUtils;
-import brooklyn.util.exceptions.RuntimeInterruptedException;
import brooklyn.util.net.Networking;
+import brooklyn.util.time.Duration;
import com.google.common.base.Throwables;
@@ -62,7 +56,7 @@ Beyond this you get blacklisted and requests may time out, or return none.
*/
@Deprecated
public static String getLocalhostExternalIp() {
- return LocalhostExternalIpLoader.getLocalhostIpWithin(Duration.of(2, TimeUnit.SECONDS));
+ return LocalhostExternalIpLoader.getLocalhostIpWithin(Duration.seconds(2));
}
/**
@@ -70,11 +64,11 @@ Beyond this you get blacklisted and requests may time out, or return none.
*/
@Deprecated
public static String getLocalhostExternalIpImpl() {
- return LocalhostExternalIpLoader.getLocalhostIpWithin(Duration.of(2, TimeUnit.SECONDS));
+ return LocalhostExternalIpLoader.getLocalhostIpWithin(Duration.seconds(2));
}
public String getLookupUrlForLocalhost() {
- return getLookupUrlForPublicIp(LocalhostExternalIpLoader.getLocalhostIp());
+ return getLookupUrlForPublicIp(LocalhostExternalIpLoader.getLocalhostIpQuicklyOrDefault());
}
/** returns URL to get properties for the given address (assuming localhost if address is on a subnet) */
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 37c68b5..483915b 100644
--- a/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
+++ b/core/src/main/java/brooklyn/management/ha/HighAvailabilityManagerImpl.java
@@ -180,16 +180,16 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
if (nodeState == ManagementNodeState.STANDBY) {
String masterNodeId = getManagementPlaneSyncState().getMasterNodeId();
ManagementNodeSyncRecord masterNodeDetails = getManagementPlaneSyncState().getManagementNodes().get(masterNodeId);
- LOG.info("Management node started as HA STANDBY autodetected, master is "+masterNodeId+
+ LOG.info("Management node "+ownNodeId+" started as HA STANDBY autodetected, master is "+masterNodeId+
(masterNodeDetails==null || masterNodeDetails.getUri()==null ? " (no further info)" : " at "+masterNodeDetails.getUri()));
} else {
- LOG.info("Management node started as HA MASTER autodetected");
+ LOG.info("Management node "+ownNodeId+" started as HA MASTER autodetected");
}
break;
case MASTER:
if (existingMaster == null) {
promoteToMaster();
- LOG.info("Management node started as HA MASTER explicitly");
+ LOG.info("Management node "+ownNodeId+" started as HA MASTER explicitly");
} else {
throw new IllegalStateException("Master already exists; cannot start as master ("+existingMaster.toVerboseString()+")");
}
@@ -197,13 +197,13 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
case STANDBY:
if (existingMaster != null) {
publishAndCheck(true);
- LOG.info("Management node started as HA STANDBY explicitly, status "+nodeState);
+ LOG.info("Management node "+ownNodeId+" started as HA STANDBY explicitly, status "+nodeState);
} else {
throw new IllegalStateException("No existing master; cannot start as standby");
}
break;
default:
- throw new IllegalStateException("Unexpected high availability start-mode "+startMode);
+ throw new IllegalStateException("Unexpected high availability start-mode "+startMode+" for "+this);
}
registerPollTask();
@@ -354,7 +354,7 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
* Looks up the state of all nodes in the management plane, and checks if the master is still ok.
* If it's not then determines which node should be promoted to master. If it is ourself, then promotes.
*/
- protected void checkMaster(boolean initializin) {
+ protected void checkMaster(boolean initializing) {
long now = currentTimeMillis();
ManagementPlaneSyncRecord memento = loadManagementPlaneSyncRecord(false);
@@ -398,7 +398,7 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager {
ownNodeMemento.toVerboseString(),
heartbeatTimeout
});
- if (!initializin) {
+ if (!initializing) {
LOG.warn("HA subsystem detected change of master from "+masterNodeId+" to "
+ (newMasterNodeId == null ? "<unknown>" :
newMasterNodeId + (newMasterNodeUri!=null ? " "+newMasterNodeUri : "")));
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/management/ha/ManagementPlaneSyncRecordPersisterToObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/ha/ManagementPlaneSyncRecordPersisterToObjectStore.java b/core/src/main/java/brooklyn/management/ha/ManagementPlaneSyncRecordPersisterToObjectStore.java
index 8969d03..9d8b202 100644
--- a/core/src/main/java/brooklyn/management/ha/ManagementPlaneSyncRecordPersisterToObjectStore.java
+++ b/core/src/main/java/brooklyn/management/ha/ManagementPlaneSyncRecordPersisterToObjectStore.java
@@ -20,6 +20,7 @@ import brooklyn.entity.rebind.persister.XmlMementoSerializer;
import brooklyn.entity.rebind.plane.dto.ManagementPlaneSyncRecordImpl;
import brooklyn.management.ManagementContext;
import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.text.Strings;
import brooklyn.util.time.Duration;
import brooklyn.util.time.Time;
@@ -67,24 +68,27 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
// TODO Leak if we go through lots of managers; but tiny!
private final ConcurrentMap<String, StoreObjectAccessorWithLock> nodeWriters = Maps.newConcurrentMap();
- private final StoreObjectAccessorWithLock masterWriter;
- private final StoreObjectAccessorWithLock changeLogWriter;
-
- private final MementoSerializer<Object> serializer;
+ private StoreObjectAccessorWithLock masterWriter;
+ private StoreObjectAccessorWithLock changeLogWriter;
+ private ManagementContext mgmt;
private final PersistenceObjectStore objectStore;
+ private final MementoSerializer<Object> serializer;
private static final int MAX_SERIALIZATION_ATTEMPTS = 5;
+ private boolean started = false;
private volatile boolean running = true;
+
/**
- * @param mgmt not used at present but handy to ensure we know it so that obj store is prepared
+ * @param mgmt not used much at present but handy to ensure we know it so that obj store is prepared
* @param objectStore the objectStore use to read/write management-plane data;
* this must have been {@link PersistenceObjectStore#prepareForUse(ManagementContext, brooklyn.entity.rebind.persister.PersistMode)}
* @param classLoader ClassLoader to use when deserializing data
*/
public ManagementPlaneSyncRecordPersisterToObjectStore(ManagementContext mgmt, PersistenceObjectStore objectStore, ClassLoader classLoader) {
+ this.mgmt = mgmt;
this.objectStore = checkNotNull(objectStore, "objectStore");
MementoSerializer<Object> rawSerializer = new XmlMementoSerializer<Object>(checkNotNull(classLoader, "classLoader"));
@@ -92,12 +96,17 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
objectStore.createSubPath(NODES_SUB_PATH);
- masterWriter = new StoreObjectAccessorLocking(objectStore.newAccessor("/master"));
- changeLogWriter = new StoreObjectAccessorLocking(objectStore.newAccessor("/change.log"));
-
LOG.debug("ManagementPlaneMemento-persister will use store "+objectStore);
}
+ protected synchronized void init() {
+ if (!started) {
+ started = true;
+ masterWriter = new StoreObjectAccessorLocking(objectStore.newAccessor("/master"));
+ changeLogWriter = new StoreObjectAccessorLocking(objectStore.newAccessor("/change.log"));
+ }
+ }
+
@Override
public void stop() {
running = false;
@@ -124,6 +133,7 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
if (!running) {
throw new IllegalStateException("Persister not running; cannot load memento from "+ objectStore.getSummaryName());
}
+ init();
// Note this is called a lot - every time we check the heartbeats
if (LOG.isTraceEnabled()) LOG.trace("Loading management-plane memento from {}", objectStore.getSummaryName());
@@ -136,7 +146,7 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
// Therefore read master file first, followed by the other node-files.
String masterNodeId = masterWriter.get();
if (masterNodeId == null) {
- LOG.warn("No entity-memento deserialized from file "+masterWriter+"; ignoring and continuing");
+ LOG.debug("No master-memento deserialized from file "+masterWriter+"; ignoring and continuing (normal on startup, should cause an error later in live operation)");
} else {
builder.masterNodeId(masterNodeId);
}
@@ -148,11 +158,29 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
for (String nodeFile : nodeFiles) {
PersistenceObjectStore.StoreObjectAccessor objectAccessor = objectStore.newAccessor(nodeFile);
- String nodeContents = objectAccessor.get();
- ManagementNodeSyncRecord memento = nodeContents==null ? null : (ManagementNodeSyncRecord) serializer.fromString(nodeContents);
+ String nodeContents = null;
+ Exception problem = null;
+ try {
+ nodeContents = objectAccessor.get();
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ problem = e;
+ }
+ if (problem!=null || Strings.isBlank(nodeContents)) {
+ // happens if node has gone away, or if FileBasedObjectStore.moveFile is not atomic,
+ // i.e. it has deleted but not updated it yet
+ if (objectAccessor.exists()) {
+ throw Exceptions.propagate(new IllegalStateException("Node record "+nodeFile+" could not be read when "+mgmt.getManagementNodeId()+" was scanning", problem));
+ } else {
+ LOG.warn("Node record "+nodeFile+" went away while "+mgmt.getManagementNodeId()+" was scanning, ignoring (it has probably been terminated)");
+ // if file was deleted, silently ignore
+ continue;
+ }
+ }
+ ManagementNodeSyncRecord memento = (ManagementNodeSyncRecord) serializer.fromString(nodeContents);
if (memento == null) {
- LOG.warn("No manager-memento deserialized from " + nodeFile + " (possibly just stopped?); ignoring" +
- " and continuing");
+ // shouldn't happen
+ throw Exceptions.propagate(new IllegalStateException("Node record "+nodeFile+" could not be deserialized when "+mgmt.getManagementNodeId()+" was scanning: "+nodeContents, problem));
} else {
builder.node(memento);
}
@@ -170,6 +198,8 @@ public class ManagementPlaneSyncRecordPersisterToObjectStore implements Manageme
if (LOG.isDebugEnabled()) LOG.debug("Persister not running; ignoring checkpointed delta of manager-memento");
return;
}
+ init();
+
if (LOG.isDebugEnabled()) LOG.debug("Checkpointed delta of manager-memento; updating {}", delta);
for (ManagementNodeSyncRecord m : delta.getNodes()) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 a93168d..c84a200 100644
--- a/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
+++ b/core/src/main/java/brooklyn/management/internal/LocalManagementContext.java
@@ -100,7 +100,7 @@ public class LocalManagementContext extends AbstractManagementContext {
private final LocalAccessManager accessManager;
private final LocalUsageManager usageManager;
- private final Throwable constructionStackTrace = new Throwable("for construction stacktrace").fillInStackTrace();
+ public final Throwable constructionStackTrace = new Throwable("for construction stacktrace").fillInStackTrace();
private final Map<String, Object> brooklynAdditionalProperties;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 1cc022a..cbe36d1 100644
--- a/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
+++ b/core/src/main/java/brooklyn/management/internal/NonDeploymentManagementContext.java
@@ -46,6 +46,8 @@ import brooklyn.mementos.BrooklynMementoPersister;
import brooklyn.util.guava.Maybe;
import brooklyn.util.time.Duration;
+import com.google.common.base.Function;
+
public class NonDeploymentManagementContext implements ManagementContextInternal {
public enum NonDeploymentManagementContextMode {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/main/java/brooklyn/util/BrooklynNetworkUtils.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/util/BrooklynNetworkUtils.java b/core/src/main/java/brooklyn/util/BrooklynNetworkUtils.java
index 3aaaa61..6274c0a 100644
--- a/core/src/main/java/brooklyn/util/BrooklynNetworkUtils.java
+++ b/core/src/main/java/brooklyn/util/BrooklynNetworkUtils.java
@@ -3,15 +3,15 @@ package brooklyn.util;
import java.net.InetAddress;
import brooklyn.config.BrooklynServiceAttributes;
-import brooklyn.location.geo.UtraceHostGeoLookup;
+import brooklyn.location.geo.LocalhostExternalIpLoader;
import brooklyn.util.flags.TypeCoercions;
import brooklyn.util.net.Networking;
public class BrooklynNetworkUtils {
- /** returns the externally-facing IP address from which this host comes */
+ /** returns the externally-facing IP address from which this host comes, or 127.0.0.1 if not resolvable */
public static String getLocalhostExternalIp() {
- return UtraceHostGeoLookup.getLocalhostExternalIp();
+ return LocalhostExternalIpLoader.getLocalhostIpQuicklyOrDefault();
}
/** returns a IP address for localhost paying attention to a system property to prevent lookup in some cases */
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/entity/basic/BrooklynShutdownHooksTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/basic/BrooklynShutdownHooksTest.java b/core/src/test/java/brooklyn/entity/basic/BrooklynShutdownHooksTest.java
index 1ff903f..fa37028 100644
--- a/core/src/test/java/brooklyn/entity/basic/BrooklynShutdownHooksTest.java
+++ b/core/src/test/java/brooklyn/entity/basic/BrooklynShutdownHooksTest.java
@@ -10,6 +10,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import brooklyn.entity.basic.BrooklynShutdownHooks.BrooklynShutdownHookJob;
import brooklyn.entity.proxying.EntitySpec;
import brooklyn.management.ManagementContext;
import brooklyn.test.entity.BlockingEntity;
@@ -38,7 +39,7 @@ public class BrooklynShutdownHooksTest {
@Test
public void testInvokeStopEntityOnShutdown() throws Exception {
BrooklynShutdownHooks.invokeStopOnShutdown(entity);
- BrooklynShutdownHooks.BrooklynShutdownHookJob job = new BrooklynShutdownHooks.BrooklynShutdownHookJob();
+ BrooklynShutdownHooks.BrooklynShutdownHookJob job = BrooklynShutdownHookJob.newInstanceForTesting();
job.run();
assertTrue(entity.getCallHistory().contains("stop"));
@@ -53,7 +54,7 @@ public class BrooklynShutdownHooksTest {
// It will timeout after shutdown-timeout
BrooklynShutdownHooks.setShutdownTimeout(Duration.of(100, TimeUnit.MILLISECONDS));
BrooklynShutdownHooks.invokeStopOnShutdown(blockingEntity);
- BrooklynShutdownHooks.BrooklynShutdownHookJob job = new BrooklynShutdownHooks.BrooklynShutdownHookJob();
+ BrooklynShutdownHooks.BrooklynShutdownHookJob job = BrooklynShutdownHookJob.newInstanceForTesting();
job.run();
latch.countDown();
@@ -62,7 +63,7 @@ public class BrooklynShutdownHooksTest {
@Test
public void testInvokeTerminateManagementContextOnShutdown() throws Exception {
BrooklynShutdownHooks.invokeTerminateOnShutdown(managementContext);
- BrooklynShutdownHooks.BrooklynShutdownHookJob job = new BrooklynShutdownHooks.BrooklynShutdownHookJob();
+ BrooklynShutdownHooks.BrooklynShutdownHookJob job = BrooklynShutdownHookJob.newInstanceForTesting();
job.run();
assertFalse(managementContext.isRunning());
@@ -73,7 +74,7 @@ public class BrooklynShutdownHooksTest {
public void testInvokeStopEntityAndTerminateManagementContextOnShutdown() throws Exception {
BrooklynShutdownHooks.invokeTerminateOnShutdown(managementContext);
BrooklynShutdownHooks.invokeStopOnShutdown(entity);
- BrooklynShutdownHooks.BrooklynShutdownHookJob job = new BrooklynShutdownHooks.BrooklynShutdownHookJob();
+ BrooklynShutdownHooks.BrooklynShutdownHookJob job = BrooklynShutdownHookJob.newInstanceForTesting();
job.run();
assertTrue(entity.getCallHistory().contains("stop"));
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/entity/group/DynamicMultiGroupTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/group/DynamicMultiGroupTest.java b/core/src/test/java/brooklyn/entity/group/DynamicMultiGroupTest.java
index 4be4d00..a7aa0d2 100644
--- a/core/src/test/java/brooklyn/entity/group/DynamicMultiGroupTest.java
+++ b/core/src/test/java/brooklyn/entity/group/DynamicMultiGroupTest.java
@@ -71,7 +71,7 @@ public class DynamicMultiGroupTest {
checkDistribution(group, dmg, childSpec, child1, child2);
}
- @Test
+ @Test(groups="Integration") // because takes 4s or so
public void testBucketDistributionWithRescan() {
Group group = app.createAndManageChild(EntitySpec.create(BasicGroup.class));
final DynamicMultiGroup dmg = app.createAndManageChild(
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 14551d2..5405f1c 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindEntityTest.java
@@ -293,11 +293,11 @@ public class RebindEntityTest extends RebindTestFixtureWithApp {
// Serialize and rebind, but don't yet manage the app
RebindTestUtils.waitForPersisted(origApp);
RebindTestUtils.checkCurrentMementoSerializable(origApp);
- newManagementContext = new LocalManagementContext();
+ newManagementContext = RebindTestUtils.newPersistingManagementContextUnstarted(mementoDir, classLoader);
Thread thread = new Thread() {
public void run() {
try {
- RebindTestUtils.rebind(newManagementContext, mementoDir, getClass().getClassLoader());
+ newManagementContext.getRebindManager().rebind(classLoader);
} catch (Exception e) {
throw Throwables.propagate(e);
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 c53983b..fa9258f 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindTestFixture.java
@@ -9,10 +9,13 @@ import brooklyn.entity.basic.Entities;
import brooklyn.entity.basic.StartableApplication;
import brooklyn.entity.rebind.persister.BrooklynMementoPersisterToObjectStore;
import brooklyn.entity.rebind.persister.FileBasedObjectStore;
+import brooklyn.entity.rebind.persister.PersistMode;
import brooklyn.internal.BrooklynFeatureEnablement;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.mementos.BrooklynMementoManifest;
+import brooklyn.test.entity.LocalManagementContextForTests;
import brooklyn.util.os.Os;
import brooklyn.util.time.Duration;
@@ -101,8 +104,11 @@ public abstract class RebindTestFixture<T extends StartableApplication> {
}
protected BrooklynMementoManifest loadMementoManifest() throws Exception {
- BrooklynMementoPersisterToObjectStore persister = new BrooklynMementoPersisterToObjectStore(
- new FileBasedObjectStore(mementoDir), classLoader);
+ newManagementContext = new LocalManagementContextForTests();
+ FileBasedObjectStore objectStore = new FileBasedObjectStore(mementoDir);
+ objectStore.injectManagementContext(newManagementContext);
+ objectStore.prepareForUse(PersistMode.AUTO, HighAvailabilityMode.DISABLED);
+ BrooklynMementoPersisterToObjectStore persister = new BrooklynMementoPersisterToObjectStore(objectStore, classLoader);
RebindExceptionHandler exceptionHandler = new RecordingRebindExceptionHandler(RebindManager.RebindFailureMode.FAIL_AT_END, RebindManager.RebindFailureMode.FAIL_AT_END);
BrooklynMementoManifest mementoManifest = persister.loadMementoManifest(exceptionHandler);
persister.stop();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 df1720a..0f8d11f 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindTestUtils.java
@@ -27,6 +27,7 @@ import brooklyn.entity.rebind.persister.PersistenceObjectStore;
import brooklyn.entity.trait.Identifiable;
import brooklyn.location.Location;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.management.internal.LocalManagementContext;
import brooklyn.mementos.BrooklynMemento;
import brooklyn.test.entity.LocalManagementContextForTests;
@@ -153,7 +154,8 @@ public class RebindTestUtils {
unstarted = new LocalManagementContextForTests();
}
- objectStore.prepareForUse(unstarted, PersistMode.AUTO);
+ objectStore.injectManagementContext(unstarted);
+ objectStore.prepareForUse(PersistMode.AUTO, HighAvailabilityMode.DISABLED);
BrooklynMementoPersisterToObjectStore newPersister = new BrooklynMementoPersisterToObjectStore(objectStore, classLoader);
((RebindManagerImpl) unstarted.getRebindManager()).setPeriodicPersistPeriod(persistPeriod);
unstarted.getRebindManager().setPersister(newPersister);
@@ -204,6 +206,8 @@ public class RebindTestUtils {
public static Application rebind(ManagementContext newManagementContext, File mementoDir, ClassLoader classLoader, RebindExceptionHandler exceptionHandler) throws Exception {
PersistenceObjectStore objectStore = new FileBasedObjectStore(mementoDir);
+ objectStore.injectManagementContext(newManagementContext);
+ objectStore.prepareForUse(PersistMode.AUTO, HighAvailabilityMode.DISABLED);
return rebind(newManagementContext, mementoDir, classLoader, exceptionHandler, objectStore);
}
public static Application rebind(ManagementContext newManagementContext, File mementoDir,
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryObjectStore.java b/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryObjectStore.java
index aea0754..12552fe 100644
--- a/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryObjectStore.java
+++ b/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryObjectStore.java
@@ -3,12 +3,11 @@ package brooklyn.entity.rebind.persister;
import java.util.List;
import java.util.Map;
-import javax.annotation.Nullable;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.util.collections.MutableList;
import brooklyn.util.collections.MutableMap;
@@ -31,12 +30,16 @@ public class InMemoryObjectStore implements PersistenceObjectStore {
}
@Override
+ public void prepareForContendedWrite() {
+ }
+
+ @Override
public void createSubPath(String subPath) {
- if (!prepared) throw new IllegalStateException("prepare method not yet invoked: "+this);
}
@Override
public StoreObjectAccessor newAccessor(final String path) {
+ if (!prepared) throw new IllegalStateException("prepare method not yet invoked: "+this);
return new StoreObjectAccessorLocking(new SingleThreadedInMemoryStoreObjectAccessor(filesByName, path));
}
@@ -86,6 +89,7 @@ public class InMemoryObjectStore implements PersistenceObjectStore {
@Override
public List<String> listContentsWithSubPath(final String parentSubPath) {
+ if (!prepared) throw new IllegalStateException("prepare method not yet invoked: "+this);
synchronized (filesByName) {
List<String> result = MutableList.of();
for (String file: filesByName.keySet())
@@ -105,7 +109,11 @@ public class InMemoryObjectStore implements PersistenceObjectStore {
}
@Override
- public void prepareForUse(ManagementContext mgmt, @Nullable PersistMode persistMode) {
+ public void injectManagementContext(ManagementContext mgmt) {
+ }
+
+ @Override
+ public void prepareForUse(PersistMode persistMode, HighAvailabilityMode haMode) {
prepared = true;
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryStoreObjectAccessorWriterTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryStoreObjectAccessorWriterTest.java b/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryStoreObjectAccessorWriterTest.java
index 86c8c91..7511ae7 100644
--- a/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryStoreObjectAccessorWriterTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/persister/InMemoryStoreObjectAccessorWriterTest.java
@@ -10,7 +10,9 @@ import brooklyn.entity.rebind.persister.PersistenceObjectStore.StoreObjectAccess
public class InMemoryStoreObjectAccessorWriterTest extends PersistenceStoreObjectAccessorWriterTestFixture {
protected StoreObjectAccessorWithLock newPersistenceStoreObjectAccessor() throws IOException {
- return new StoreObjectAccessorLocking(new InMemoryObjectStore().newAccessor("foo"));
+ InMemoryObjectStore store = new InMemoryObjectStore();
+ store.prepareForUse(null, null);
+ return new StoreObjectAccessorLocking(store.newAccessor("foo"));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/entity/rebind/persister/ListeningObjectStore.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/persister/ListeningObjectStore.java b/core/src/test/java/brooklyn/entity/rebind/persister/ListeningObjectStore.java
index 37e3cbb..a478b46 100644
--- a/core/src/test/java/brooklyn/entity/rebind/persister/ListeningObjectStore.java
+++ b/core/src/test/java/brooklyn/entity/rebind/persister/ListeningObjectStore.java
@@ -9,6 +9,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.util.text.Strings;
import brooklyn.util.time.CountdownTimer;
import brooklyn.util.time.Duration;
@@ -105,19 +106,28 @@ public class ListeningObjectStore implements PersistenceObjectStore {
this.listener = Preconditions.checkNotNull(listener);
}
+ @Override
public String getSummaryName() {
return delegate.getSummaryName();
}
+ @Override
+ public void prepareForContendedWrite() {
+ delegate.prepareForContendedWrite();
+ }
+
+ @Override
public StoreObjectAccessor newAccessor(String path) {
return new ListeningAccessor(path, delegate.newAccessor(path));
}
+ @Override
public void createSubPath(String subPath) {
listener.recordQueryOut("creating path "+subPath, 1+subPath.length());
delegate.createSubPath(subPath);
}
+ @Override
public List<String> listContentsWithSubPath(String subPath) {
listener.recordQueryOut("requesting list "+subPath, 1+subPath.length());
List<String> result = delegate.listContentsWithSubPath(subPath);
@@ -125,14 +135,22 @@ public class ListeningObjectStore implements PersistenceObjectStore {
return result;
}
+ @Override
public void close() {
delegate.close();
}
- public void prepareForUse(ManagementContext managementContext, PersistMode persistMode) {
- delegate.prepareForUse(managementContext, persistMode);
+ @Override
+ public void injectManagementContext(ManagementContext managementContext) {
+ delegate.injectManagementContext(managementContext);
+ }
+
+ @Override
+ public void prepareForUse(PersistMode persistMode, HighAvailabilityMode haMode) {
+ delegate.prepareForUse(persistMode, haMode);
}
+ @Override
public void deleteCompletely() {
listener.recordDataOut("deleting completely", 1);
delegate.deleteCompletely();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/core/src/test/java/brooklyn/location/geo/LocalhostExternalIpLoaderIntegrationTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/location/geo/LocalhostExternalIpLoaderIntegrationTest.java b/core/src/test/java/brooklyn/location/geo/LocalhostExternalIpLoaderIntegrationTest.java
index dd2efa1..3c01834 100644
--- a/core/src/test/java/brooklyn/location/geo/LocalhostExternalIpLoaderIntegrationTest.java
+++ b/core/src/test/java/brooklyn/location/geo/LocalhostExternalIpLoaderIntegrationTest.java
@@ -29,7 +29,7 @@ public class LocalhostExternalIpLoaderIntegrationTest {
@Test(groups = "Integration")
public void testLoadExternalIp() {
- assertNotNull(LocalhostExternalIpLoader.getLocalhostIp());
+ assertNotNull(LocalhostExternalIpLoader.getLocalhostIpWaiting());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/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 959c0ba..ee005e0 100644
--- a/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
+++ b/core/src/test/java/brooklyn/management/ha/HighAvailabilityManagerTestFixture.java
@@ -60,7 +60,8 @@ public abstract class HighAvailabilityManagerTestFixture {
managementContext = newLocalManagementContext();
ownNodeId = managementContext.getManagementNodeId();
objectStore = newPersistenceObjectStore();
- objectStore.prepareForUse(managementContext, PersistMode.CLEAN);
+ objectStore.injectManagementContext(managementContext);
+ objectStore.prepareForUse(PersistMode.CLEAN, HighAvailabilityMode.DISABLED);
persister = new ManagementPlaneSyncRecordPersisterToObjectStore(managementContext, objectStore, classLoader);
BrooklynMementoPersisterToObjectStore persisterObj = new BrooklynMementoPersisterToObjectStore(objectStore, classLoader);
managementContext.getRebindManager().setPersister(persisterObj);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/locations/jclouds/src/main/java/brooklyn/entity/rebind/persister/jclouds/JcloudsBlobStoreBasedObjectStore.java
----------------------------------------------------------------------
diff --git a/locations/jclouds/src/main/java/brooklyn/entity/rebind/persister/jclouds/JcloudsBlobStoreBasedObjectStore.java b/locations/jclouds/src/main/java/brooklyn/entity/rebind/persister/jclouds/JcloudsBlobStoreBasedObjectStore.java
index 6893847..a91bca0 100644
--- a/locations/jclouds/src/main/java/brooklyn/entity/rebind/persister/jclouds/JcloudsBlobStoreBasedObjectStore.java
+++ b/locations/jclouds/src/main/java/brooklyn/entity/rebind/persister/jclouds/JcloudsBlobStoreBasedObjectStore.java
@@ -25,6 +25,7 @@ import brooklyn.location.basic.LocationConfigKeys;
import brooklyn.location.cloud.CloudLocationConfig;
import brooklyn.location.jclouds.JcloudsLocation;
import brooklyn.management.ManagementContext;
+import brooklyn.management.ha.HighAvailabilityMode;
import brooklyn.util.exceptions.FatalConfigurationRuntimeException;
import brooklyn.util.text.Strings;
@@ -43,7 +44,6 @@ public class JcloudsBlobStoreBasedObjectStore implements PersistenceObjectStore
private ManagementContext mgmt;
-
public JcloudsBlobStoreBasedObjectStore(String locationSpec, String containerName) {
this.locationSpec = locationSpec;
this.containerName = containerName;
@@ -86,13 +86,16 @@ public class JcloudsBlobStoreBasedObjectStore implements PersistenceObjectStore
return context;
}
+ @Override
+ public void prepareForContendedWrite() {
+ }
+
public String getContainerName() {
return containerName;
}
@Override
public void createSubPath(String subPath) {
- checkPrepared();
// not needed, and buggy on softlayer w swift w jclouds 1.7.2:
// throws a "not found" if we're creating an empty directory from scratch
// context.getBlobStore().createDirectory(getContainerName(), subPath);
@@ -152,11 +155,16 @@ public class JcloudsBlobStoreBasedObjectStore implements PersistenceObjectStore
}
@Override
- public void prepareForUse(ManagementContext mgmt, @Nullable PersistMode persistMode) {
+ public void injectManagementContext(ManagementContext mgmt) {
if (this.mgmt!=null && !this.mgmt.equals(mgmt))
throw new IllegalStateException("Cannot change mgmt context of "+this);
this.mgmt = mgmt;
-
+ }
+
+ @Override
+ public void prepareForUse(@Nullable PersistMode persistMode, HighAvailabilityMode haMode) {
+ if (mgmt==null) throw new NullPointerException("Must inject ManagementContext before preparing "+this);
+
getBlobStoreContext();
if (persistMode==null || persistMode==PersistMode.DISABLED) {
@@ -165,24 +173,9 @@ public class JcloudsBlobStoreBasedObjectStore implements PersistenceObjectStore
}
Boolean backups = mgmt.getConfig().getConfig(BrooklynServerConfig.PERSISTENCE_BACKUPS_REQUIRED);
- if (backups==null) backups = false;
- if (backups) {
+ if (Boolean.TRUE.equals(backups)) {
throw new FatalConfigurationRuntimeException("Backups not supported for object store ("+this+")");
}
-
- switch (persistMode) {
- case CLEAN:
- deleteCompletely();
- break;
- case REBIND:
- // no op - could confirm it exists (?)
- break;
- case AUTO:
- // again, no op
- break;
- default:
- throw new FatalConfigurationRuntimeException("Unexpected persist mode "+persistMode+"; modified during initialization?!");
- }
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/3c444b47/locations/jclouds/src/main/java/brooklyn/location/jclouds/networking/JcloudsLocationSecurityGroupCustomizer.java
----------------------------------------------------------------------
diff --git a/locations/jclouds/src/main/java/brooklyn/location/jclouds/networking/JcloudsLocationSecurityGroupCustomizer.java b/locations/jclouds/src/main/java/brooklyn/location/jclouds/networking/JcloudsLocationSecurityGroupCustomizer.java
index ed4e567..cb861a9 100644
--- a/locations/jclouds/src/main/java/brooklyn/location/jclouds/networking/JcloudsLocationSecurityGroupCustomizer.java
+++ b/locations/jclouds/src/main/java/brooklyn/location/jclouds/networking/JcloudsLocationSecurityGroupCustomizer.java
@@ -17,12 +17,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import brooklyn.entity.Entity;
-import brooklyn.location.geo.UtraceHostGeoLookup;
+import brooklyn.location.geo.LocalhostExternalIpLoader;
import brooklyn.location.jclouds.BasicJcloudsLocationCustomizer;
import brooklyn.location.jclouds.JcloudsLocation;
import brooklyn.location.jclouds.JcloudsSshMachineLocation;
import brooklyn.util.net.Cidr;
import brooklyn.util.task.Tasks;
+import brooklyn.util.time.Duration;
import com.google.common.annotations.Beta;
import com.google.common.annotations.VisibleForTesting;
@@ -360,7 +361,7 @@ public final class JcloudsLocationSecurityGroupCustomizer extends BasicJcloudsLo
synchronized (this) {
local = cidr;
if (local == null) {
- String externalIp = UtraceHostGeoLookup.getLocalhostExternalIp();
+ String externalIp = LocalhostExternalIpLoader.getLocalhostIpWithin(Duration.seconds(5));
cidr = local = new Cidr(externalIp + "/32");
}
}