X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fclustering%2Fservices_implementation%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fclustering%2Fservices_implementation%2Finternal%2FClusterManager.java;h=cd3a29579194c9a6a9831f4e44d7e67150b69f6d;hp=d3cd23e35015723d6bbac1bda92b756844a0a920;hb=81004658f4171a0f06c3a5496f39c61801189c04;hpb=42210c03b0a4c54706320ba9f55794c0abd4d201 diff --git a/opendaylight/clustering/services_implementation/src/main/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusterManager.java b/opendaylight/clustering/services_implementation/src/main/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusterManager.java index d3cd23e350..cd3a295791 100644 --- a/opendaylight/clustering/services_implementation/src/main/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusterManager.java +++ b/opendaylight/clustering/services_implementation/src/main/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusterManager.java @@ -9,8 +9,6 @@ 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; @@ -35,6 +33,9 @@ import javax.transaction.TransactionManager; 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; @@ -54,15 +55,16 @@ import org.opendaylight.controller.clustering.services.IClusterServices; 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 roleChangeListeners; private ViewChangedListener cacheManagerListener; private static String loopbackAddress = "127.0.0.1"; @@ -97,12 +99,12 @@ public class ClusterManager implements IClusterServices { if (supernodes.hasMoreTokens()) { // Populate the list of my addresses try { - Enumeration e = NetworkInterface.getNetworkInterfaces(); + Enumeration e = NetworkInterface.getNetworkInterfaces(); while (e.hasMoreElements()) { - NetworkInterface n = (NetworkInterface) e.nextElement(); - Enumeration ee = n.getInetAddresses(); + NetworkInterface n = e.nextElement(); + Enumeration ee = n.getInetAddresses(); while (ee.hasMoreElements()) { - InetAddress i = (InetAddress) ee.nextElement(); + InetAddress i = ee.nextElement(); myAddresses.add(i); } } @@ -113,15 +115,15 @@ public class ClusterManager implements IClusterServices { } 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(); @@ -184,13 +186,13 @@ public class ClusterManager implements IClusterServices { 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; } } @@ -199,7 +201,7 @@ public class ClusterManager implements IClusterServices { .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 @@ -207,8 +209,8 @@ public class ClusterManager implements IClusterServices { 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 @@ -220,15 +222,16 @@ public class ClusterManager implements IClusterServices { // 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; } @@ -241,18 +244,20 @@ public class ClusterManager implements IClusterServices { 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(); @@ -260,15 +265,13 @@ public class ClusterManager implements IClusterServices { 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() { @@ -289,7 +292,7 @@ public class ClusterManager implements IClusterServices { String cacheName, Set cMode) throws CacheExistException, CacheConfigException { EmbeddedCacheManager manager = this.cm; - Cache c; + Cache c; String realCacheName = "{" + containerName + "}_{" + cacheName + "}"; if (manager == null) { return null; @@ -322,7 +325,7 @@ public class ClusterManager implements IClusterServices { @Override public ConcurrentMap getCache(String containerName, String cacheName) { EmbeddedCacheManager manager = this.cm; - Cache c; + Cache c; String realCacheName = "{" + containerName + "}_{" + cacheName + "}"; if (manager == null) { return null; @@ -359,7 +362,7 @@ public class ClusterManager implements IClusterServices { @Override public Set getCacheList(String containerName) { - Set perContainerCaches = new HashSet(); + Set perContainerCaches = new HashSet(); EmbeddedCacheManager manager = this.cm; if (manager == null) { return null; @@ -403,7 +406,7 @@ public class ClusterManager implements IClusterServices { public void addListener(String containerName, String cacheName, IGetUpdates u) throws CacheListenerAddException { EmbeddedCacheManager manager = this.cm; - Cache c; + Cache c; String realCacheName = "{" + containerName + "}_{" + cacheName + "}"; if (manager == null) { return; @@ -415,9 +418,6 @@ public class ClusterManager implements IClusterServices { c = manager.getCache(realCacheName); CacheListenerContainer cl = new CacheListenerContainer(u, containerName, cacheName); - if (cl == null) { - throw new CacheListenerAddException(); - } c.addListener(cl); } @@ -425,7 +425,7 @@ public class ClusterManager implements IClusterServices { public Set> getListeners(String containerName, String cacheName) { EmbeddedCacheManager manager = this.cm; - Cache c; + Cache c; String realCacheName = "{" + containerName + "}_{" + cacheName + "}"; if (manager == null) { return null; @@ -436,7 +436,7 @@ public class ClusterManager implements IClusterServices { } c = manager.getCache(realCacheName); - Set> res = new HashSet(); + Set> res = new HashSet>(); Set listeners = c.getListeners(); for (Object listener : listeners) { if (listener instanceof CacheListenerContainer) { @@ -452,7 +452,7 @@ public class ClusterManager implements IClusterServices { public void removeListener(String containerName, String cacheName, IGetUpdates u) { EmbeddedCacheManager manager = this.cm; - Cache c; + Cache c; String realCacheName = "{" + containerName + "}_{" + cacheName + "}"; if (manager == null) { return; @@ -539,7 +539,7 @@ public class ClusterManager implements IClusterServices { 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()); @@ -549,7 +549,7 @@ public class ClusterManager implements IClusterServices { 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(); @@ -571,25 +571,29 @@ public class ClusterManager implements IClusterServices { return null; } + @Override public List getClusteredControllers() { EmbeddedCacheManager manager = this.cm; if (manager == null) { return null; } List
controllers = manager.getMembers(); - if ((controllers == null) || controllers.size() == 0) + if ((controllers == null) || controllers.size() == 0) { return null; + } List clusteredControllers = new ArrayList(); 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) { @@ -603,7 +607,7 @@ public class ClusterManager implements IClusterServices { 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; } @@ -616,12 +620,12 @@ public class ClusterManager implements IClusterServices { 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(); this.cacheManagerListener = new ViewChangedListener( this.roleChangeListeners); manager.addListener(this.cacheManagerListener); @@ -637,7 +641,7 @@ public class ClusterManager implements IClusterServices { 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; } @@ -669,4 +673,21 @@ public class ClusterManager implements IClusterServices { } } } + + 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); + } }