package org.opendaylight.controller.clustering.services_implementation.internal;
-import java.io.PrintWriter;
-import java.io.StringWriter;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import org.infinispan.Cache;
import org.infinispan.configuration.cache.Configuration;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
+import org.infinispan.configuration.parsing.ParserRegistry;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.notifications.Listener;
import org.opendaylight.controller.clustering.services.IGetUpdates;
import org.opendaylight.controller.clustering.services.IListenRoleChange;
import org.opendaylight.controller.clustering.services.ListenRoleChangeAddException;
+import org.opendaylight.controller.sal.core.IContainerAware;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ClusterManager implements IClusterServices {
+public class ClusterManager implements IClusterServices, IContainerAware {
protected static final Logger logger = LoggerFactory
.getLogger(ClusterManager.class);
private DefaultCacheManager cm;
GossipRouter gossiper;
- private HashSet roleChangeListeners;
+ private HashSet<IListenRoleChange> roleChangeListeners;
private ViewChangedListener cacheManagerListener;
private static String loopbackAddress = "127.0.0.1";
if (supernodes.hasMoreTokens()) {
// Populate the list of my addresses
try {
- Enumeration e = NetworkInterface.getNetworkInterfaces();
+ Enumeration<NetworkInterface> e = NetworkInterface.getNetworkInterfaces();
while (e.hasMoreElements()) {
- NetworkInterface n = (NetworkInterface) e.nextElement();
- Enumeration ee = n.getInetAddresses();
+ NetworkInterface n = e.nextElement();
+ Enumeration<InetAddress> ee = n.getInetAddresses();
while (ee.hasMoreElements()) {
- InetAddress i = (InetAddress) ee.nextElement();
+ InetAddress i = ee.nextElement();
myAddresses.add(i);
}
}
}
while (supernodes.hasMoreTokens()) {
String curr_supernode = supernodes.nextToken();
- logger.debug("Examining supernode " + curr_supernode);
+ logger.debug("Examining supernode {}", curr_supernode);
StringTokenizer host_port = new StringTokenizer(curr_supernode,
"[]");
String host;
String port;
Integer port_num = gossipRouterPortDefault;
if (host_port.countTokens() > 2) {
- logger.error("Error parsing supernode " + curr_supernode
- + " proceed to the next one");
+ logger.error("Error parsing supernode {} proceed to the next one",
+ curr_supernode);
continue;
}
host = host_port.nextToken();
for (InetAddress myAddr : myAddresses) {
if (myAddr.isLoopbackAddress()
|| myAddr.isLinkLocalAddress()) {
- logger.debug("Skipping local address "
- + myAddr.getHostAddress());
+ logger.debug("Skipping local address {}",
+ myAddr.getHostAddress());
continue;
} else {
// First non-local address
myBind = myAddr.getHostAddress();
- logger.debug("First non-local address " + myBind);
+ logger.debug("First non-local address {}", myBind);
break;
}
}
.getProperty("jgroups.tcp.address");
if (jgroupAddress == null) {
if (myBind != null) {
- logger.debug("Set bind address to be " + myBind);
+ logger.debug("Set bind address to be {}", myBind);
System.setProperty("jgroups.tcp.address", myBind);
} else {
logger
System.setProperty("jgroups.tcp.address", "127.0.0.1");
}
} else {
- logger.debug("jgroup.tcp.address already set to be "
- + jgroupAddress);
+ logger.debug("jgroup.tcp.address already set to be {}",
+ jgroupAddress);
}
} catch (UnknownHostException uhe) {
logger
// host list
System.setProperty("jgroups.tcpgossip.initial_hosts",
sanitized_supernodes_list.toString());
- logger.debug("jgroups.tcp.address set to "
- + System.getProperty("jgroups.tcp.address"));
- logger.debug("jgroups.tcpgossip.initial_hosts set to "
- + System.getProperty("jgroups.tcpgossip.initial_hosts"));
+ logger.debug("jgroups.tcp.address set to {}",
+ System.getProperty("jgroups.tcp.address"));
+ logger.debug("jgroups.tcpgossip.initial_hosts set to {}",
+ System.getProperty("jgroups.tcpgossip.initial_hosts"));
GossipRouter res = null;
if (amIGossipRouter) {
- logger.info("I'm a GossipRouter will listen on port "
- + gossipRouterPort);
- res = new GossipRouter(gossipRouterPort);
+ logger.info("I'm a GossipRouter will listen on port {}",
+ gossipRouterPort);
+ // Start a GossipRouter with JMX support
+ res = new GossipRouter(gossipRouterPort, null, true);
}
return res;
}
this.gossiper.start();
logger.info("Started GossipRouter");
} catch (Exception e) {
- logger.error("GossipRouter didn't start exception " + e
- + " met");
- StringWriter sw = new StringWriter();
- logger.error("Stack Trace that raised the exception");
- e.printStackTrace(new PrintWriter(sw));
- logger.error(sw.toString());
+ logger.error("GossipRouter didn't start. Exception Stack Trace",
+ e);
}
}
logger.info("Starting the ClusterManager");
try {
- //FIXME keeps throwing FileNotFoundException
- this.cm = new DefaultCacheManager("/config/infinispan-config.xml");
+ ParserRegistry parser = new ParserRegistry(this.getClass()
+ .getClassLoader());
+ ConfigurationBuilderHolder holder = parser.parseFile("config/infinispan-config.xml");
+ GlobalConfigurationBuilder globalBuilder = holder.getGlobalConfigurationBuilder();
+ globalBuilder.serialization()
+ .classResolver(new ClassResolver())
+ .build();
+ this.cm = new DefaultCacheManager(holder, false);
logger.debug("Allocated ClusterManager");
if (this.cm != null) {
this.cm.start();
logger.debug("Started the ClusterManager");
}
} catch (Exception ioe) {
- StringWriter sw = new StringWriter();
logger.error("Cannot configure infinispan .. bailing out ");
logger.error("Stack Trace that raised th exception");
- ioe.printStackTrace(new PrintWriter(sw));
- logger.error(sw.toString());
+ logger.error("",ioe);
this.cm = null;
this.stop();
}
- logger.debug("Cache Manager has value " + this.cm);
+ logger.debug("Cache Manager has value {}", this.cm);
}
public void stop() {
String cacheName, Set<cacheMode> cMode) throws CacheExistException,
CacheConfigException {
EmbeddedCacheManager manager = this.cm;
- Cache c;
+ Cache<Object,Object> c;
String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
if (manager == null) {
return null;
@Override
public ConcurrentMap<?, ?> getCache(String containerName, String cacheName) {
EmbeddedCacheManager manager = this.cm;
- Cache c;
+ Cache<Object,Object> c;
String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
if (manager == null) {
return null;
@Override
public Set<String> getCacheList(String containerName) {
- Set<String> perContainerCaches = new HashSet();
+ Set<String> perContainerCaches = new HashSet<String>();
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
return null;
public void addListener(String containerName, String cacheName,
IGetUpdates<?, ?> u) throws CacheListenerAddException {
EmbeddedCacheManager manager = this.cm;
- Cache c;
+ Cache<Object,Object> c;
String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
if (manager == null) {
return;
public Set<IGetUpdates<?, ?>> getListeners(String containerName,
String cacheName) {
EmbeddedCacheManager manager = this.cm;
- Cache c;
+ Cache<Object,Object> c;
String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
if (manager == null) {
return null;
}
c = manager.getCache(realCacheName);
- Set<IGetUpdates<?, ?>> res = new HashSet();
+ Set<IGetUpdates<?, ?>> res = new HashSet<IGetUpdates<?, ?>>();
Set<Object> listeners = c.getListeners();
for (Object listener : listeners) {
if (listener instanceof CacheListenerContainer) {
public void removeListener(String containerName, String cacheName,
IGetUpdates<?, ?> u) {
EmbeddedCacheManager manager = this.cm;
- Cache c;
+ Cache<Object,Object> c;
String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
if (manager == null) {
return;
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
// In case we cannot fetch the information, lets assume we
- // are standby, so to have less responsability.
+ // are standby, so to have less responsibility.
return true;
}
return (!manager.isCoordinator());
EmbeddedCacheManager manager = this.cm;
if ((manager == null) || (a == null)) {
// In case we cannot fetch the information, lets assume we
- // are standby, so to have less responsability.
+ // are standby, so to have less responsibility.
return null;
}
Transport t = manager.getTransport();
return null;
}
+ @Override
public List<InetAddress> getClusteredControllers() {
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
return null;
}
List<Address> controllers = manager.getMembers();
- if ((controllers == null) || controllers.size() == 0)
+ if ((controllers == null) || controllers.size() == 0) {
return null;
+ }
List<InetAddress> clusteredControllers = new ArrayList<InetAddress>();
for (Address a : controllers) {
InetAddress inetAddress = addressToInetAddress(a);
if (inetAddress != null
- && !inetAddress.getHostAddress().equals(loopbackAddress))
+ && !inetAddress.getHostAddress().equals(loopbackAddress)) {
clusteredControllers.add(inetAddress);
+ }
}
return clusteredControllers;
}
+ @Override
public InetAddress getMyAddress() {
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
// In case we cannot fetch the information, lets assume we
- // are standby, so to have less responsability.
+ // are standby, so to have less responsibility.
return null;
}
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
// In case we cannot fetch the information, lets assume we
- // are standby, so to have less responsability.
+ // are standby, so to have less responsibility.
throw new ListenRoleChangeAddException();
}
if (this.roleChangeListeners == null) {
- this.roleChangeListeners = new HashSet();
+ this.roleChangeListeners = new HashSet<IListenRoleChange>();
this.cacheManagerListener = new ViewChangedListener(
this.roleChangeListeners);
manager.addListener(this.cacheManagerListener);
EmbeddedCacheManager manager = this.cm;
if (manager == null) {
// In case we cannot fetch the information, lets assume we
- // are standby, so to have less responsability.
+ // are standby, so to have less responsibility.
return;
}
}
}
}
+
+ private void removeContainerCaches(String containerName) {
+ logger.info("Destroying caches for container {}", containerName);
+ for (String cacheName : this.getCacheList(containerName)) {
+ this.destroyCache(containerName, cacheName);
+ }
+ }
+
+ @Override
+ public void containerCreate(String arg0) {
+ // no op
+ }
+
+ @Override
+ public void containerDestroy(String container) {
+ removeContainerCaches(container);
+ }
}
Code Review / controller.git / blobdiff