ensembleProperties.put("dataDir", options.getZooKeeperServerDataDir() + File.separator + newClusterId);
// create new ensemble
String ensembleProfileId = "fabric-ensemble-" + newClusterId;
IllegalStateAssertion.assertFalse(profileRegistry.get().hasProfile(versionId, ensembleProfileId), "Profile already exists: " + versionId + "/" + ensembleProfileId);
ProfileBuilder ensembleProfileBuilder = ProfileBuilder.Factory.create(versionId, ensembleProfileId);
ensembleProfileBuilder.addAttribute(Profile.ABSTRACT, "true").addAttribute(Profile.HIDDEN, "true");
int index = 1;
String connectionUrl = "";
String realConnectionUrl = "";
String containerList = "";
List<Profile> memberProfiles = new ArrayList<>();
for (String container : containers) {
String ip = getSubstitutedPath(curator.get(), ZkPath.CONTAINER_IP.getPath(container));
String minimumPort = String.valueOf(Ports.MIN_PORT_NUMBER);
String maximumPort = String.valueOf(Ports.MAX_PORT_NUMBER);
String bindAddress = "0.0.0.0";
if (exists(curator.get(), ZkPath.CONTAINER_PORT_MIN.getPath(container)) != null) {
minimumPort = getSubstitutedPath(curator.get(), ZkPath.CONTAINER_PORT_MIN.getPath(container));
}
if (exists(curator.get(), ZkPath.CONTAINER_PORT_MAX.getPath(container)) != null) {
maximumPort = getSubstitutedPath(curator.get(), ZkPath.CONTAINER_PORT_MAX.getPath(container));
}
if (exists(curator.get(), ZkPath.CONTAINER_BINDADDRESS.getPath(container)) != null) {
bindAddress = getSubstitutedPath(curator.get(), ZkPath.CONTAINER_BINDADDRESS.getPath(container));
}
// Ensemble member properties
Properties memberProperties = new Properties();
String memberPropertiesName = "io.fabric8.zookeeper.server-" + newClusterId + ".properties";
String port1 = publicPort(container, Integer.toString(findPort(usedPorts, ip, mapPortToRange(Ports.DEFAULT_ZOOKEEPER_SERVER_PORT, minimumPort, maximumPort))));
if (containers.size() > 1) {
String port2 = publicPort(container, Integer.toString(findPort(usedPorts, ip, mapPortToRange(Ports.DEFAULT_ZOOKEEPER_PEER_PORT, minimumPort, maximumPort))));
String port3 = publicPort(container, Integer.toString(findPort(usedPorts, ip, mapPortToRange(Ports.DEFAULT_ZOOKEEPER_ELECTION_PORT, minimumPort, maximumPort))));
ensembleProperties.put("server." + Integer.toString(index), "${zk:" + container + "/ip}:" + port2 + ":" + port3);
memberProperties.put("server.id", Integer.toString(index));
}
memberProperties.put("clientPort", port1);
memberProperties.put("clientPortAddress", bindAddress);
// Create ensemble member profile
String memberProfileId = "fabric-ensemble-" + newClusterId + "-" + index;
IllegalStateAssertion.assertFalse(profileRegistry.get().hasProfile(versionId, memberProfileId), "Profile already exists: " + versionId + "/" + memberProfileId);
ProfileBuilder memberProfileBuilder = ProfileBuilder.Factory.create(versionId, memberProfileId);
memberProfileBuilder.addAttribute(Profile.HIDDEN, "true").addAttribute(Profile.PARENTS, ensembleProfileId);
memberProfileBuilder.addFileConfiguration(memberPropertiesName, DataStoreUtils.toBytes(memberProperties));
memberProfiles.add(memberProfileBuilder.getProfile());
if (connectionUrl.length() > 0) {
connectionUrl += ",";
realConnectionUrl += ",";
}
connectionUrl += "${zk:" + container + "/ip}:" + port1;
realConnectionUrl += ip + ":" + port1;
if (containerList.length() > 0) {
containerList += ",";
}
containerList += container;
index++;
}
LockHandle writeLock = profileRegistry.get().aquireWriteLock();
try {
// Create the ensemble profile
ensembleProfileBuilder.addFileConfiguration(ensemblePropertiesName, DataStoreUtils.toBytes(ensembleProperties));
Profile ensembleProfile = ensembleProfileBuilder.getProfile();
LOGGER.info("Creating parent ensemble profile: {}", ensembleProfile);
profileRegistry.get().createProfile(ensembleProfile);
// Create the member profiles
for (Profile memberProfile : memberProfiles) {
LOGGER.info("Creating member ensemble profile: {}", memberProfile);
profileRegistry.get().createProfile(memberProfile);
}
} finally {
writeLock.unlock();
}
index = 1;
for (String container : containers) {
// add this container to the ensemble
List<String> profiles = new LinkedList<String>(dataStore.get().getContainerProfiles(container));
profiles.add("fabric-ensemble-" + newClusterId + "-" + Integer.toString(index));
LOGGER.info("Assigning member ensemble profile with id: {} to {}.", ensembleProfileId + "-" + index, container);
dataStore.get().setContainerProfiles(container, profiles);
index++;
}
Profile defaultProfile = profileRegistry.get().getRequiredProfile(versionId, "default");
Map<String, String> zkConfig = defaultProfile.getConfiguration(Constants.ZOOKEEPER_CLIENT_PID);
if (oldClusterId != null) {
Properties properties = DataStoreUtils.toProperties(zkConfig);
properties.put("zookeeper.url", getSubstitutedData(curator.get(), realConnectionUrl));
properties.put("zookeeper.password", options.getZookeeperPassword());
CuratorFramework dst = CuratorFrameworkFactory.builder().connectString(realConnectionUrl).retryPolicy(new RetryOneTime(500))
.aclProvider(aclProvider.get()).authorization("digest", ("fabric:" + options.getZookeeperPassword()).getBytes()).sessionTimeoutMs(30000)
.connectionTimeoutMs((int) options.getMigrationTimeout()).build();
dst.start();
try {
long t0 = System.currentTimeMillis();
LOGGER.info("Waiting for ensemble {} to become ready.", newClusterId);
if (!dst.getZookeeperClient().blockUntilConnectedOrTimedOut()) {
throw new EnsembleModificationFailed("Timed out connecting to new ensemble.", EnsembleModificationFailed.Reason.TIMEOUT);
}
LOGGER.info("Copying data from the old ensemble to the new one");
copy(curator.get(), dst, "/fabric");
setData(dst, ZkPath.CONFIG_ENSEMBLES.getPath(), newClusterId);
setData(dst, ZkPath.CONFIG_ENSEMBLE.getPath(newClusterId), containerList);
// Perform cleanup when the new datastore has been registered.
final AtomicReference<DataStore> result = new AtomicReference<DataStore>();
runtimeProperties.get().putRuntimeAttribute(DataStoreTemplate.class, new DataStoreTemplate() {
@Override
public void doWith(ProfileRegistry profileRegistry, DataStore dataStore) {
synchronized (result) {
result.set(dataStore);
result.notifyAll();
}
}
});
LOGGER.info("Migrating containers to the new ensemble using url {}.", connectionUrl);
setData(dst, ZkPath.CONFIG_ENSEMBLE_PASSWORD.getPath(), PasswordEncoder.encode(options.getZookeeperPassword()));
setData(dst, ZkPath.CONFIG_ENSEMBLE_URL.getPath(), connectionUrl);
curator.get().inTransaction()
.setData().forPath(ZkPath.CONFIG_ENSEMBLE_PASSWORD.getPath(), PasswordEncoder.encode(options.getZookeeperPassword()).getBytes(Charsets.UTF_8))
.and()
.setData().forPath(ZkPath.CONFIG_ENSEMBLE_URL.getPath(), connectionUrl.getBytes(Charsets.UTF_8))
.and().commit();
// Wait until all containers switched
boolean allStarted = false;
while (!allStarted && System.currentTimeMillis() - t0 < options.getMigrationTimeout()) {
allStarted = true;
for (Container container : allContainers) {
allStarted &= exists(dst, ZkPath.CONTAINER_ALIVE.getPath(container.getId())) != null;
}
if (!allStarted) {
Thread.sleep(1000);
}
}
if (!allStarted) {
throw new EnsembleModificationFailed("Timeout waiting for containers to join the new ensemble", EnsembleModificationFailed.Reason.TIMEOUT);
}
LOGGER.info("Migration successful. Cleaning up");
// Wait until the new datastore has been registered
synchronized (result) {
if (result.get() == null) {
result.wait();
}
}
// Remove old profiles
for (String container : oldContainers) {
cleanUpEnsembleProfiles(result.get(), container, oldClusterId);
}
} finally {
dst.close();
}
} else {
ProfileBuilder builder = ProfileBuilder.Factory.createFrom(defaultProfile);
zkConfig = new HashMap<>(zkConfig);
zkConfig.put("zookeeper.password", "${zk:" + ZkPath.CONFIG_ENSEMBLE_PASSWORD.getPath() + "}");
zkConfig.put("zookeeper.url", "${zk:" + ZkPath.CONFIG_ENSEMBLE_URL.getPath() + "}");
builder.addConfiguration(Constants.ZOOKEEPER_CLIENT_PID, zkConfig);
profileRegistry.get().updateProfile(builder.getProfile());
}
} catch (Exception e) {
throw EnsembleModificationFailed.launderThrowable(e);
}
}