Merge "Declare a property for commons-lang version in poms and use it."
[controller.git] / opendaylight / clustering / services_implementation / src / main / java / org / opendaylight / controller / clustering / services_implementation / internal / ClusterManager.java
1
2 /*
3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
4  *
5  * This program and the accompanying materials are made available under the
6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
7  * and is available at http://www.eclipse.org/legal/epl-v10.html
8  */
9
10 package org.opendaylight.controller.clustering.services_implementation.internal;
11
12 import java.net.InetAddress;
13 import java.net.NetworkInterface;
14 import java.net.SocketException;
15 import java.net.UnknownHostException;
16 import java.util.ArrayList;
17 import java.util.EnumSet;
18 import java.util.Enumeration;
19 import java.util.HashSet;
20 import java.util.List;
21 import java.util.Properties;
22 import java.util.Set;
23 import java.util.StringTokenizer;
24 import java.util.concurrent.ConcurrentMap;
25
26 import javax.transaction.HeuristicMixedException;
27 import javax.transaction.HeuristicRollbackException;
28 import javax.transaction.NotSupportedException;
29 import javax.transaction.RollbackException;
30 import javax.transaction.SystemException;
31 import javax.transaction.Transaction;
32 import javax.transaction.TransactionManager;
33
34 import org.infinispan.Cache;
35 import org.infinispan.configuration.cache.Configuration;
36 import org.infinispan.configuration.global.GlobalConfigurationBuilder;
37 import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
38 import org.infinispan.configuration.parsing.ParserRegistry;
39 import org.infinispan.manager.DefaultCacheManager;
40 import org.infinispan.manager.EmbeddedCacheManager;
41 import org.infinispan.notifications.Listener;
42 import org.infinispan.notifications.cachemanagerlistener.annotation.ViewChanged;
43 import org.infinispan.notifications.cachemanagerlistener.event.ViewChangedEvent;
44 import org.infinispan.remoting.transport.Address;
45 import org.infinispan.remoting.transport.Transport;
46 import org.infinispan.remoting.transport.jgroups.JGroupsAddress;
47 import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
48 import org.jgroups.Channel;
49 import org.jgroups.Event;
50 import org.jgroups.stack.GossipRouter;
51 import org.opendaylight.controller.clustering.services.CacheConfigException;
52 import org.opendaylight.controller.clustering.services.CacheExistException;
53 import org.opendaylight.controller.clustering.services.CacheListenerAddException;
54 import org.opendaylight.controller.clustering.services.IClusterServices;
55 import org.opendaylight.controller.clustering.services.IGetUpdates;
56 import org.opendaylight.controller.clustering.services.IListenRoleChange;
57 import org.opendaylight.controller.clustering.services.ListenRoleChangeAddException;
58 import org.opendaylight.controller.sal.core.IContainerAware;
59 import org.slf4j.Logger;
60 import org.slf4j.LoggerFactory;
61
62 public class ClusterManager implements IClusterServices, IContainerAware {
63     protected static final Logger logger = LoggerFactory
64             .getLogger(ClusterManager.class);
65     private DefaultCacheManager cm;
66     GossipRouter gossiper;
67     private HashSet<IListenRoleChange> roleChangeListeners;
68     private ViewChangedListener cacheManagerListener;
69
70     private static String loopbackAddress = "127.0.0.1";
71
72     /**
73      * Start a JGroups GossipRouter if we are a supernode. The
74      * GosispRouter is nothing more than a simple
75      * rendevouz-pointer. All the nodes that wants to join the cluster
76      * will come to any of the rendevouz point and they introduce the
77      * nodes to all the others. Once the meet and greet phase if over,
78      * the nodes will open a full-mesh with the remaining n-1 nodes,
79      * so even if the GossipRouter goes down nothing is lost.
80      * NOTE: This function has the side effect to set some of the
81      * JGROUPS configurations, this because in this function already
82      * we try to retrieve some of the network capabilities of the
83      * device and so it's better not to do that again
84      *
85      *
86      * @return GossipRouter
87      */
88     private GossipRouter startGossiper() {
89         boolean amIGossipRouter = false;
90         Integer gossipRouterPortDefault = 12001;
91         Integer gossipRouterPort = gossipRouterPortDefault;
92         InetAddress gossipRouterAddress = null;
93         String supernodes_list = System.getProperty("supernodes",
94                 loopbackAddress);
95         StringBuffer sanitized_supernodes_list = new StringBuffer();
96         List<InetAddress> myAddresses = new ArrayList<InetAddress>();
97
98         StringTokenizer supernodes = new StringTokenizer(supernodes_list, ":");
99         if (supernodes.hasMoreTokens()) {
100             // Populate the list of my addresses
101             try {
102                 Enumeration<NetworkInterface> e = NetworkInterface.getNetworkInterfaces();
103                 while (e.hasMoreElements()) {
104                     NetworkInterface n = e.nextElement();
105                     Enumeration<InetAddress> ee = n.getInetAddresses();
106                     while (ee.hasMoreElements()) {
107                         InetAddress i = ee.nextElement();
108                         myAddresses.add(i);
109                     }
110                 }
111             } catch (SocketException se) {
112                 logger.error("Cannot get the list of network interfaces");
113                 return null;
114             }
115         }
116         while (supernodes.hasMoreTokens()) {
117             String curr_supernode = supernodes.nextToken();
118             logger.debug("Examining supernode {}", curr_supernode);
119             StringTokenizer host_port = new StringTokenizer(curr_supernode,
120                     "[]");
121             String host;
122             String port;
123             Integer port_num = gossipRouterPortDefault;
124             if (host_port.countTokens() > 2) {
125                 logger.error("Error parsing supernode {} proceed to the next one",
126                         curr_supernode);
127                 continue;
128             }
129             host = host_port.nextToken();
130             InetAddress hostAddr;
131             try {
132                 hostAddr = InetAddress.getByName(host);
133             } catch (UnknownHostException ue) {
134                 logger.error("Host not known");
135                 continue;
136             }
137             if (host_port.hasMoreTokens()) {
138                 port = host_port.nextToken();
139                 try {
140                     port_num = Integer.valueOf(port);
141                 } catch (NumberFormatException ne) {
142                     logger
143                             .error("Supplied supernode gossiepr port is not recognized, using standard gossipport");
144                     port_num = gossipRouterPortDefault;
145                 }
146                 if ((port_num > 65535) || (port_num < 0)) {
147                     logger
148                             .error("Supplied supernode gossip port is outside a valid TCP port range");
149                     port_num = gossipRouterPortDefault;
150                 }
151             }
152             if (!amIGossipRouter) {
153                 if (host != null) {
154                     for (InetAddress myAddr : myAddresses) {
155                         if (myAddr.equals(hostAddr)) {
156                             amIGossipRouter = true;
157                             gossipRouterAddress = hostAddr;
158                             gossipRouterPort = port_num;
159                             break;
160                         }
161                     }
162                 }
163             }
164             if (!sanitized_supernodes_list.toString().equals("")) {
165                 sanitized_supernodes_list.append(",");
166             }
167             sanitized_supernodes_list.append(hostAddr.getHostAddress() + "["
168                     + port_num + "]");
169         }
170
171         if (amIGossipRouter) {
172             // Set the Jgroups binding interface to the one we got
173             // from the supernodes attribute
174             if (gossipRouterAddress != null) {
175                 System.setProperty("jgroups.tcp.address", gossipRouterAddress
176                         .getHostAddress());
177             }
178         } else {
179             // Set the Jgroup binding interface to the one we are well
180             // known outside or else to the first with non-local
181             // scope.
182             try {
183                 String myBind = InetAddress.getLocalHost().getHostAddress();
184                 if (myBind == null
185                         || InetAddress.getLocalHost().isLoopbackAddress()) {
186                     for (InetAddress myAddr : myAddresses) {
187                         if (myAddr.isLoopbackAddress()
188                                 || myAddr.isLinkLocalAddress()) {
189                             logger.debug("Skipping local address {}",
190                                          myAddr.getHostAddress());
191                             continue;
192                         } else {
193                             // First non-local address
194                             myBind = myAddr.getHostAddress();
195                             logger.debug("First non-local address {}", myBind);
196                             break;
197                         }
198                     }
199                 }
200                 String jgroupAddress = System
201                         .getProperty("jgroups.tcp.address");
202                 if (jgroupAddress == null) {
203                     if (myBind != null) {
204                         logger.debug("Set bind address to be {}", myBind);
205                         System.setProperty("jgroups.tcp.address", myBind);
206                     } else {
207                         logger
208                                 .debug("Set bind address to be LOCALHOST=127.0.0.1");
209                         System.setProperty("jgroups.tcp.address", "127.0.0.1");
210                     }
211                 } else {
212                     logger.debug("jgroup.tcp.address already set to be {}",
213                             jgroupAddress);
214                 }
215             } catch (UnknownHostException uhe) {
216                 logger
217                         .error("Met UnknownHostException while trying to get binding address for jgroups");
218             }
219         }
220
221         // The supernodes list constitute also the tcpgossip initial
222         // host list
223         System.setProperty("jgroups.tcpgossip.initial_hosts",
224                 sanitized_supernodes_list.toString());
225         logger.debug("jgroups.tcp.address set to {}",
226                 System.getProperty("jgroups.tcp.address"));
227         logger.debug("jgroups.tcpgossip.initial_hosts set to {}",
228                 System.getProperty("jgroups.tcpgossip.initial_hosts"));
229         GossipRouter res = null;
230         if (amIGossipRouter) {
231             logger.info("I'm a GossipRouter will listen on port {}",
232                     gossipRouterPort);
233             // Start a GossipRouter with JMX support
234             res = new GossipRouter(gossipRouterPort, null, true);
235         }
236         return res;
237     }
238
239     public void start() {
240         this.gossiper = startGossiper();
241         if (this.gossiper != null) {
242             logger.debug("Trying to start Gossiper");
243             try {
244                 this.gossiper.start();
245                 logger.info("Started GossipRouter");
246             } catch (Exception e) {
247                 logger.error("GossipRouter didn't start. Exception Stack Trace",
248                              e);
249             }
250         }
251         logger.info("Starting the ClusterManager");
252         try {
253             ParserRegistry parser = new ParserRegistry(this.getClass()
254                     .getClassLoader());
255             ConfigurationBuilderHolder holder = parser.parseFile("config/infinispan-config.xml");
256             GlobalConfigurationBuilder globalBuilder = holder.getGlobalConfigurationBuilder();
257             globalBuilder.serialization()
258                     .classResolver(new ClassResolver())
259                     .build();
260             this.cm = new DefaultCacheManager(holder, false);
261             logger.debug("Allocated ClusterManager");
262             if (this.cm != null) {
263                 this.cm.start();
264                 this.cm.startCache();
265                 logger.debug("Started the ClusterManager");
266             }
267         } catch (Exception ioe) {
268             logger.error("Cannot configure infinispan .. bailing out ");
269             logger.error("Stack Trace that raised th exception");
270             logger.error("",ioe);
271             this.cm = null;
272             this.stop();
273         }
274         logger.debug("Cache Manager has value {}", this.cm);
275     }
276
277     public void stop() {
278         logger.info("Stopping the ClusterManager");
279         if (this.cm != null) {
280             logger.info("Found a valid ClusterManager, now let it be stopped");
281             this.cm.stop();
282             this.cm = null;
283         }
284         if (this.gossiper != null) {
285             this.gossiper.stop();
286             this.gossiper = null;
287         }
288     }
289
290     @Override
291     public ConcurrentMap<?, ?> createCache(String containerName,
292             String cacheName, Set<cacheMode> cMode) throws CacheExistException,
293             CacheConfigException {
294         EmbeddedCacheManager manager = this.cm;
295         Cache<Object,Object> c;
296         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
297         if (manager == null) {
298             return null;
299         }
300
301         if (manager.cacheExists(realCacheName)) {
302             throw new CacheExistException();
303         }
304
305         // Sanity check to avoid contrasting parameters
306         if (cMode.containsAll(EnumSet.of(
307                 IClusterServices.cacheMode.NON_TRANSACTIONAL,
308                 IClusterServices.cacheMode.TRANSACTIONAL))) {
309             throw new CacheConfigException();
310         }
311
312         if (cMode.contains(IClusterServices.cacheMode.NON_TRANSACTIONAL)) {
313             c = manager.getCache(realCacheName);
314             return c;
315         } else if (cMode.contains(IClusterServices.cacheMode.TRANSACTIONAL)) {
316             Configuration rc = manager
317                     .getCacheConfiguration("transactional-type");
318             manager.defineConfiguration(realCacheName, rc);
319             c = manager.getCache(realCacheName);
320             return c;
321         }
322         return null;
323     }
324
325     @Override
326     public ConcurrentMap<?, ?> getCache(String containerName, String cacheName) {
327         EmbeddedCacheManager manager = this.cm;
328         Cache<Object,Object> c;
329         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
330         if (manager == null) {
331             return null;
332         }
333
334         if (manager.cacheExists(realCacheName)) {
335             c = manager.getCache(realCacheName);
336             return c;
337         }
338         return null;
339     }
340
341     @Override
342     public void destroyCache(String containerName, String cacheName) {
343         EmbeddedCacheManager manager = this.cm;
344         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
345         if (manager == null) {
346             return;
347         }
348         if (manager.cacheExists(realCacheName)) {
349             manager.removeCache(realCacheName);
350         }
351     }
352
353     @Override
354     public boolean existCache(String containerName, String cacheName) {
355         EmbeddedCacheManager manager = this.cm;
356         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
357         if (manager == null) {
358             return false;
359         }
360         return manager.cacheExists(realCacheName);
361     }
362
363     @Override
364     public Set<String> getCacheList(String containerName) {
365         Set<String> perContainerCaches = new HashSet<String>();
366         EmbeddedCacheManager manager = this.cm;
367         if (manager == null) {
368             return null;
369         }
370         for (String cacheName : manager.getCacheNames()) {
371             if (cacheName.startsWith("{" + containerName + "}_")) {
372                 String[] res = cacheName.split("[{}]");
373                 if (res.length >= 4 && res[1].equals(containerName)
374                         && res[2].equals("_")) {
375                     perContainerCaches.add(res[3]);
376                 }
377             }
378         }
379
380         return (perContainerCaches);
381     }
382
383     @Override
384     public Properties getCacheProperties(String containerName, String cacheName) {
385         EmbeddedCacheManager manager = this.cm;
386         if (manager == null) {
387             return null;
388         }
389         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
390         if (!manager.cacheExists(realCacheName)) {
391             return null;
392         }
393         Configuration conf = manager.getCache(realCacheName).getAdvancedCache()
394                 .getCacheConfiguration();
395         Properties p = new Properties();
396         p.setProperty(IClusterServices.cacheProps.TRANSACTION_PROP.toString(),
397                 conf.transaction().toString());
398         p.setProperty(IClusterServices.cacheProps.CLUSTERING_PROP.toString(),
399                 conf.clustering().toString());
400         p.setProperty(IClusterServices.cacheProps.LOCKING_PROP.toString(), conf
401                 .locking().toString());
402         return p;
403     }
404
405     @Override
406     public void addListener(String containerName, String cacheName,
407             IGetUpdates<?, ?> u) throws CacheListenerAddException {
408         EmbeddedCacheManager manager = this.cm;
409         Cache<Object,Object> c;
410         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
411         if (manager == null) {
412             return;
413         }
414
415         if (!manager.cacheExists(realCacheName)) {
416             throw new CacheListenerAddException();
417         }
418         c = manager.getCache(realCacheName);
419         CacheListenerContainer cl = new CacheListenerContainer(u,
420                 containerName, cacheName);
421         c.addListener(cl);
422     }
423
424     @Override
425     public Set<IGetUpdates<?, ?>> getListeners(String containerName,
426             String cacheName) {
427         EmbeddedCacheManager manager = this.cm;
428         Cache<Object,Object> c;
429         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
430         if (manager == null) {
431             return null;
432         }
433
434         if (!manager.cacheExists(realCacheName)) {
435             return null;
436         }
437         c = manager.getCache(realCacheName);
438
439         Set<IGetUpdates<?, ?>> res = new HashSet<IGetUpdates<?, ?>>();
440         Set<Object> listeners = c.getListeners();
441         for (Object listener : listeners) {
442             if (listener instanceof CacheListenerContainer) {
443                 CacheListenerContainer cl = (CacheListenerContainer) listener;
444                 res.add(cl.whichListener());
445             }
446         }
447
448         return res;
449     }
450
451     @Override
452     public void removeListener(String containerName, String cacheName,
453             IGetUpdates<?, ?> u) {
454         EmbeddedCacheManager manager = this.cm;
455         Cache<Object,Object> c;
456         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
457         if (manager == null) {
458             return;
459         }
460
461         if (!manager.cacheExists(realCacheName)) {
462             return;
463         }
464         c = manager.getCache(realCacheName);
465
466         Set<Object> listeners = c.getListeners();
467         for (Object listener : listeners) {
468             if (listener instanceof CacheListenerContainer) {
469                 CacheListenerContainer cl = (CacheListenerContainer) listener;
470                 if (cl.whichListener() == u) {
471                     c.removeListener(listener);
472                     return;
473                 }
474             }
475         }
476     }
477
478     @Override
479     public void tbegin() throws NotSupportedException, SystemException {
480         EmbeddedCacheManager manager = this.cm;
481         if (manager == null) {
482             throw new IllegalStateException();
483         }
484         TransactionManager tm = manager.getCache("transactional-type")
485                 .getAdvancedCache().getTransactionManager();
486         if (tm == null) {
487             throw new IllegalStateException();
488         }
489         tm.begin();
490     }
491
492     @Override
493     public void tcommit() throws RollbackException, HeuristicMixedException,
494             HeuristicRollbackException, java.lang.SecurityException,
495             java.lang.IllegalStateException, SystemException {
496         EmbeddedCacheManager manager = this.cm;
497         if (manager == null) {
498             throw new IllegalStateException();
499         }
500         TransactionManager tm = manager.getCache("transactional-type")
501                 .getAdvancedCache().getTransactionManager();
502         if (tm == null) {
503             throw new IllegalStateException();
504         }
505         tm.commit();
506     }
507
508     @Override
509     public void trollback() throws java.lang.IllegalStateException,
510             java.lang.SecurityException, SystemException {
511         EmbeddedCacheManager manager = this.cm;
512         if (manager == null) {
513             throw new IllegalStateException();
514         }
515         TransactionManager tm = manager.getCache("transactional-type")
516                 .getAdvancedCache().getTransactionManager();
517         if (tm == null) {
518             throw new IllegalStateException();
519         }
520         tm.rollback();
521     }
522
523     @Override
524     public Transaction tgetTransaction() throws SystemException {
525         EmbeddedCacheManager manager = this.cm;
526         if (manager == null) {
527             throw new IllegalStateException();
528         }
529         TransactionManager tm = manager.getCache("transactional-type")
530                 .getAdvancedCache().getTransactionManager();
531         if (tm == null) {
532             return null;
533         }
534         return tm.getTransaction();
535     }
536
537     @Override
538     public boolean amIStandby() {
539         EmbeddedCacheManager manager = this.cm;
540         if (manager == null) {
541             // In case we cannot fetch the information, lets assume we
542             // are standby, so to have less responsibility.
543             return true;
544         }
545         return (!manager.isCoordinator());
546     }
547
548     private InetAddress addressToInetAddress(Address a) {
549         EmbeddedCacheManager manager = this.cm;
550         if ((manager == null) || (a == null)) {
551             // In case we cannot fetch the information, lets assume we
552             // are standby, so to have less responsibility.
553             return null;
554         }
555         Transport t = manager.getTransport();
556         if (t instanceof JGroupsTransport) {
557             JGroupsTransport jt = (JGroupsTransport) t;
558             Channel c = jt.getChannel();
559             if (a instanceof JGroupsAddress) {
560                 JGroupsAddress ja = (JGroupsAddress) a;
561                 org.jgroups.Address phys = (org.jgroups.Address) c
562                         .down(new Event(Event.GET_PHYSICAL_ADDRESS, ja
563                                 .getJGroupsAddress()));
564                 if (phys instanceof org.jgroups.stack.IpAddress) {
565                     InetAddress bindAddress = ((org.jgroups.stack.IpAddress) phys)
566                             .getIpAddress();
567                     return bindAddress;
568                 }
569             }
570         }
571         return null;
572     }
573
574     @Override
575     public List<InetAddress> getClusteredControllers() {
576         EmbeddedCacheManager manager = this.cm;
577         if (manager == null) {
578             return null;
579         }
580         List<Address> controllers = manager.getMembers();
581         if ((controllers == null) || controllers.size() == 0) {
582             return null;
583         }
584
585         List<InetAddress> clusteredControllers = new ArrayList<InetAddress>();
586         for (Address a : controllers) {
587             InetAddress inetAddress = addressToInetAddress(a);
588             if (inetAddress != null
589                     && !inetAddress.getHostAddress().equals(loopbackAddress)) {
590                 clusteredControllers.add(inetAddress);
591             }
592         }
593         return clusteredControllers;
594     }
595
596     @Override
597     public InetAddress getMyAddress() {
598         EmbeddedCacheManager manager = this.cm;
599         if (manager == null) {
600             return null;
601         }
602         return addressToInetAddress(manager.getAddress());
603     }
604
605     @Override
606     public InetAddress getActiveAddress() {
607         EmbeddedCacheManager manager = this.cm;
608         if (manager == null) {
609             // In case we cannot fetch the information, lets assume we
610             // are standby, so to have less responsibility.
611             return null;
612         }
613
614         return addressToInetAddress(manager.getCoordinator());
615     }
616
617     @Override
618     public void listenRoleChange(IListenRoleChange i)
619             throws ListenRoleChangeAddException {
620         EmbeddedCacheManager manager = this.cm;
621         if (manager == null) {
622             // In case we cannot fetch the information, lets assume we
623             // are standby, so to have less responsibility.
624             throw new ListenRoleChangeAddException();
625         }
626
627         if (this.roleChangeListeners == null) {
628             this.roleChangeListeners = new HashSet<IListenRoleChange>();
629             this.cacheManagerListener = new ViewChangedListener(
630                     this.roleChangeListeners);
631             manager.addListener(this.cacheManagerListener);
632         }
633
634         if (this.roleChangeListeners != null) {
635             this.roleChangeListeners.add(i);
636         }
637     }
638
639     @Override
640     public void unlistenRoleChange(IListenRoleChange i) {
641         EmbeddedCacheManager manager = this.cm;
642         if (manager == null) {
643             // In case we cannot fetch the information, lets assume we
644             // are standby, so to have less responsibility.
645             return;
646         }
647
648         if (this.roleChangeListeners != null) {
649             this.roleChangeListeners.remove(i);
650         }
651
652         if ((this.roleChangeListeners != null && this.roleChangeListeners
653                 .isEmpty())
654                 && (this.cacheManagerListener != null)) {
655             manager.removeListener(this.cacheManagerListener);
656             this.cacheManagerListener = null;
657             this.roleChangeListeners = null;
658         }
659     }
660
661     @Listener
662     public class ViewChangedListener {
663         Set<IListenRoleChange> roleListeners;
664
665         public ViewChangedListener(Set<IListenRoleChange> s) {
666             this.roleListeners = s;
667         }
668
669         @ViewChanged
670         public void viewChanged(ViewChangedEvent e) {
671             for (IListenRoleChange i : this.roleListeners) {
672                 i.newActiveAvailable();
673             }
674         }
675     }
676
677     private void removeContainerCaches(String containerName) {
678         logger.info("Destroying caches for container {}", containerName);
679         for (String cacheName : this.getCacheList(containerName)) {
680             this.destroyCache(containerName, cacheName);
681         }
682     }
683
684     @Override
685     public void containerCreate(String arg0) {
686         // no op
687     }
688
689     @Override
690     public void containerDestroy(String container) {
691         removeContainerCaches(container);
692     }
693 }