Added a trivial implementation of the LeadershipService.
Also renamed SdnIpLeadershipService to HazelcastLeadershipService, and moved it into the distributed core bundle. This allows applications which depend on LeadershipService to be used with the trivial core. Change-Id: Ie71a946d95653a4d2209afd3af0e7f23b5a4f818
Showing
4 changed files
with
108 additions
and
52 deletions
... | @@ -29,18 +29,15 @@ import org.onlab.onos.cluster.ClusterService; | ... | @@ -29,18 +29,15 @@ import org.onlab.onos.cluster.ClusterService; |
29 | import org.onlab.onos.cluster.ControllerNode; | 29 | import org.onlab.onos.cluster.ControllerNode; |
30 | import org.onlab.onos.cluster.LeadershipEvent; | 30 | import org.onlab.onos.cluster.LeadershipEvent; |
31 | import org.onlab.onos.cluster.LeadershipEventListener; | 31 | import org.onlab.onos.cluster.LeadershipEventListener; |
32 | -// import org.onlab.onos.cluster.LeadershipService; | 32 | +import org.onlab.onos.cluster.LeadershipService; |
33 | import org.onlab.onos.core.ApplicationId; | 33 | import org.onlab.onos.core.ApplicationId; |
34 | import org.onlab.onos.core.CoreService; | 34 | import org.onlab.onos.core.CoreService; |
35 | -import org.onlab.onos.event.EventDeliveryService; | ||
36 | import org.onlab.onos.net.host.HostService; | 35 | import org.onlab.onos.net.host.HostService; |
37 | import org.onlab.onos.net.intent.IntentService; | 36 | import org.onlab.onos.net.intent.IntentService; |
38 | import org.onlab.onos.sdnip.bgp.BgpRouteEntry; | 37 | import org.onlab.onos.sdnip.bgp.BgpRouteEntry; |
39 | import org.onlab.onos.sdnip.bgp.BgpSession; | 38 | import org.onlab.onos.sdnip.bgp.BgpSession; |
40 | import org.onlab.onos.sdnip.bgp.BgpSessionManager; | 39 | import org.onlab.onos.sdnip.bgp.BgpSessionManager; |
41 | import org.onlab.onos.sdnip.config.SdnIpConfigReader; | 40 | import org.onlab.onos.sdnip.config.SdnIpConfigReader; |
42 | -import org.onlab.onos.store.hz.StoreService; | ||
43 | - | ||
44 | import org.slf4j.Logger; | 41 | import org.slf4j.Logger; |
45 | 42 | ||
46 | /** | 43 | /** |
... | @@ -70,13 +67,7 @@ public class SdnIp implements SdnIpService { | ... | @@ -70,13 +67,7 @@ public class SdnIp implements SdnIpService { |
70 | protected ClusterService clusterService; | 67 | protected ClusterService clusterService; |
71 | 68 | ||
72 | @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | 69 | @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) |
73 | - protected StoreService storeService; | 70 | + protected LeadershipService leadershipService; |
74 | - | ||
75 | - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
76 | - protected EventDeliveryService eventDispatcher; | ||
77 | - | ||
78 | - // @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
79 | - protected SdnIpLeadershipService leadershipService; | ||
80 | 71 | ||
81 | private IntentSynchronizer intentSynchronizer; | 72 | private IntentSynchronizer intentSynchronizer; |
82 | private SdnIpConfigReader config; | 73 | private SdnIpConfigReader config; |
... | @@ -114,10 +105,6 @@ public class SdnIp implements SdnIpService { | ... | @@ -114,10 +105,6 @@ public class SdnIp implements SdnIpService { |
114 | interfaceService, hostService); | 105 | interfaceService, hostService); |
115 | router.start(); | 106 | router.start(); |
116 | 107 | ||
117 | - leadershipService = new SdnIpLeadershipService(clusterService, | ||
118 | - storeService, | ||
119 | - eventDispatcher); | ||
120 | - leadershipService.start(); | ||
121 | leadershipService.addListener(leadershipEventListener); | 108 | leadershipService.addListener(leadershipEventListener); |
122 | leadershipService.runForLeadership(appId.name()); | 109 | leadershipService.runForLeadership(appId.name()); |
123 | 110 | ||
... | @@ -138,7 +125,6 @@ public class SdnIp implements SdnIpService { | ... | @@ -138,7 +125,6 @@ public class SdnIp implements SdnIpService { |
138 | 125 | ||
139 | leadershipService.withdraw(appId.name()); | 126 | leadershipService.withdraw(appId.name()); |
140 | leadershipService.removeListener(leadershipEventListener); | 127 | leadershipService.removeListener(leadershipEventListener); |
141 | - leadershipService.stop(); | ||
142 | 128 | ||
143 | log.info("SDN-IP Stopped"); | 129 | log.info("SDN-IP Stopped"); |
144 | } | 130 | } | ... | ... |
... | @@ -13,7 +13,10 @@ | ... | @@ -13,7 +13,10 @@ |
13 | * See the License for the specific language governing permissions and | 13 | * See the License for the specific language governing permissions and |
14 | * limitations under the License. | 14 | * limitations under the License. |
15 | */ | 15 | */ |
16 | -package org.onlab.onos.sdnip; | 16 | +package org.onlab.onos.store.cluster.impl; |
17 | + | ||
18 | +import static com.google.common.base.Preconditions.checkArgument; | ||
19 | +import static org.onlab.util.Tools.namedThreads; | ||
17 | 20 | ||
18 | import java.util.Map; | 21 | import java.util.Map; |
19 | import java.util.concurrent.ExecutorService; | 22 | import java.util.concurrent.ExecutorService; |
... | @@ -21,6 +24,12 @@ import java.util.concurrent.Executors; | ... | @@ -21,6 +24,12 @@ import java.util.concurrent.Executors; |
21 | import java.util.concurrent.Future; | 24 | import java.util.concurrent.Future; |
22 | import java.util.concurrent.locks.Lock; | 25 | import java.util.concurrent.locks.Lock; |
23 | 26 | ||
27 | +import org.apache.felix.scr.annotations.Activate; | ||
28 | +import org.apache.felix.scr.annotations.Component; | ||
29 | +import org.apache.felix.scr.annotations.Deactivate; | ||
30 | +import org.apache.felix.scr.annotations.Reference; | ||
31 | +import org.apache.felix.scr.annotations.ReferenceCardinality; | ||
32 | +import org.apache.felix.scr.annotations.Service; | ||
24 | import org.onlab.onos.cluster.ClusterService; | 33 | import org.onlab.onos.cluster.ClusterService; |
25 | import org.onlab.onos.cluster.ControllerNode; | 34 | import org.onlab.onos.cluster.ControllerNode; |
26 | import org.onlab.onos.cluster.Leadership; | 35 | import org.onlab.onos.cluster.Leadership; |
... | @@ -34,18 +43,15 @@ import org.onlab.onos.store.hz.StoreService; | ... | @@ -34,18 +43,15 @@ import org.onlab.onos.store.hz.StoreService; |
34 | import org.onlab.onos.store.serializers.KryoNamespaces; | 43 | import org.onlab.onos.store.serializers.KryoNamespaces; |
35 | import org.onlab.onos.store.serializers.KryoSerializer; | 44 | import org.onlab.onos.store.serializers.KryoSerializer; |
36 | import org.onlab.util.KryoNamespace; | 45 | import org.onlab.util.KryoNamespace; |
37 | -import static org.onlab.util.Tools.namedThreads; | ||
38 | - | ||
39 | import org.slf4j.Logger; | 46 | import org.slf4j.Logger; |
40 | import org.slf4j.LoggerFactory; | 47 | import org.slf4j.LoggerFactory; |
48 | + | ||
41 | import com.google.common.collect.Maps; | 49 | import com.google.common.collect.Maps; |
42 | import com.hazelcast.config.TopicConfig; | 50 | import com.hazelcast.config.TopicConfig; |
43 | import com.hazelcast.core.ITopic; | 51 | import com.hazelcast.core.ITopic; |
44 | import com.hazelcast.core.Message; | 52 | import com.hazelcast.core.Message; |
45 | import com.hazelcast.core.MessageListener; | 53 | import com.hazelcast.core.MessageListener; |
46 | 54 | ||
47 | -import static com.google.common.base.Preconditions.checkArgument; | ||
48 | - | ||
49 | /** | 55 | /** |
50 | * Distributed implementation of LeadershipService that is based on Hazelcast. | 56 | * Distributed implementation of LeadershipService that is based on Hazelcast. |
51 | * <p> | 57 | * <p> |
... | @@ -63,9 +69,11 @@ import static com.google.common.base.Preconditions.checkArgument; | ... | @@ -63,9 +69,11 @@ import static com.google.common.base.Preconditions.checkArgument; |
63 | * the current leader (e.g., for informational purpose). | 69 | * the current leader (e.g., for informational purpose). |
64 | * </p> | 70 | * </p> |
65 | */ | 71 | */ |
66 | -public class SdnIpLeadershipService implements LeadershipService { | 72 | +@Component(immediate = true) |
73 | +@Service | ||
74 | +public class HazelcastLeadershipService implements LeadershipService { | ||
67 | private static final Logger log = | 75 | private static final Logger log = |
68 | - LoggerFactory.getLogger(SdnIpLeadershipService.class); | 76 | + LoggerFactory.getLogger(HazelcastLeadershipService.class); |
69 | 77 | ||
70 | private static final KryoSerializer SERIALIZER = new KryoSerializer() { | 78 | private static final KryoSerializer SERIALIZER = new KryoSerializer() { |
71 | @Override | 79 | @Override |
... | @@ -80,45 +88,31 @@ public class SdnIpLeadershipService implements LeadershipService { | ... | @@ -80,45 +88,31 @@ public class SdnIpLeadershipService implements LeadershipService { |
80 | private static final long LEADERSHIP_PERIODIC_INTERVAL_MS = 5 * 1000; // 5s | 88 | private static final long LEADERSHIP_PERIODIC_INTERVAL_MS = 5 * 1000; // 5s |
81 | private static final long LEADERSHIP_REMOTE_TIMEOUT_MS = 15 * 1000; // 15s | 89 | private static final long LEADERSHIP_REMOTE_TIMEOUT_MS = 15 * 1000; // 15s |
82 | 90 | ||
83 | - private ClusterService clusterService; | 91 | + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) |
84 | - private StoreService storeService; | 92 | + protected ClusterService clusterService; |
85 | - private EventDeliveryService eventDispatcher; | 93 | + |
94 | + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
95 | + protected StoreService storeService; | ||
96 | + | ||
97 | + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
98 | + protected EventDeliveryService eventDispatcher; | ||
86 | 99 | ||
87 | private AbstractListenerRegistry<LeadershipEvent, LeadershipEventListener> | 100 | private AbstractListenerRegistry<LeadershipEvent, LeadershipEventListener> |
88 | listenerRegistry; | 101 | listenerRegistry; |
89 | private final Map<String, Topic> topics = Maps.newConcurrentMap(); | 102 | private final Map<String, Topic> topics = Maps.newConcurrentMap(); |
90 | private ControllerNode localNode; | 103 | private ControllerNode localNode; |
91 | 104 | ||
92 | - /** | 105 | + @Activate |
93 | - * Constructor. | 106 | + protected void activate() { |
94 | - * | ||
95 | - * @param clusterService the cluster service to use | ||
96 | - * @param storeService the store service to use | ||
97 | - * @param eventDispatcher the event dispacher to use | ||
98 | - */ | ||
99 | - SdnIpLeadershipService(ClusterService clusterService, | ||
100 | - StoreService storeService, | ||
101 | - EventDeliveryService eventDispatcher) { | ||
102 | - this.clusterService = clusterService; | ||
103 | - this.storeService = storeService; | ||
104 | - this.eventDispatcher = eventDispatcher; | ||
105 | - } | ||
106 | - | ||
107 | - /** | ||
108 | - * Starts operation. | ||
109 | - */ | ||
110 | - void start() { | ||
111 | localNode = clusterService.getLocalNode(); | 107 | localNode = clusterService.getLocalNode(); |
112 | listenerRegistry = new AbstractListenerRegistry<>(); | 108 | listenerRegistry = new AbstractListenerRegistry<>(); |
113 | eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry); | 109 | eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry); |
114 | 110 | ||
115 | - log.info("SDN-IP Leadership Service started"); | 111 | + log.info("Hazelcast Leadership Service started"); |
116 | } | 112 | } |
117 | 113 | ||
118 | - /** | 114 | + @Deactivate |
119 | - * Stops operation. | 115 | + protected void deactivate() { |
120 | - */ | ||
121 | - void stop() { | ||
122 | eventDispatcher.removeSink(LeadershipEvent.class); | 116 | eventDispatcher.removeSink(LeadershipEvent.class); |
123 | 117 | ||
124 | for (Topic topic : topics.values()) { | 118 | for (Topic topic : topics.values()) { |
... | @@ -126,7 +120,7 @@ public class SdnIpLeadershipService implements LeadershipService { | ... | @@ -126,7 +120,7 @@ public class SdnIpLeadershipService implements LeadershipService { |
126 | } | 120 | } |
127 | topics.clear(); | 121 | topics.clear(); |
128 | 122 | ||
129 | - log.info("SDN-IP Leadership Service stopped"); | 123 | + log.info("Hazelcast Leadership Service stopped"); |
130 | } | 124 | } |
131 | 125 | ||
132 | @Override | 126 | @Override | ... | ... |
... | @@ -43,7 +43,7 @@ import com.google.common.collect.Sets; | ... | @@ -43,7 +43,7 @@ import com.google.common.collect.Sets; |
43 | * Distributed implementation of LeadershipService that is based on the primitives exposed by | 43 | * Distributed implementation of LeadershipService that is based on the primitives exposed by |
44 | * LockService. | 44 | * LockService. |
45 | */ | 45 | */ |
46 | -@Component(immediate = true) | 46 | +@Component(enabled = false) |
47 | @Service | 47 | @Service |
48 | public class LeadershipManager implements LeadershipService { | 48 | public class LeadershipManager implements LeadershipService { |
49 | 49 | ... | ... |
core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleLeadershipManager.java
0 → 100644
1 | +package org.onlab.onos.store.trivial.impl; | ||
2 | + | ||
3 | +import java.util.Map; | ||
4 | +import java.util.Set; | ||
5 | +import java.util.concurrent.ConcurrentHashMap; | ||
6 | +import java.util.concurrent.CopyOnWriteArraySet; | ||
7 | + | ||
8 | +import org.apache.felix.scr.annotations.Component; | ||
9 | +import org.apache.felix.scr.annotations.Reference; | ||
10 | +import org.apache.felix.scr.annotations.ReferenceCardinality; | ||
11 | +import org.apache.felix.scr.annotations.Service; | ||
12 | +import org.onlab.onos.cluster.ClusterService; | ||
13 | +import org.onlab.onos.cluster.ControllerNode; | ||
14 | +import org.onlab.onos.cluster.Leadership; | ||
15 | +import org.onlab.onos.cluster.LeadershipEvent; | ||
16 | +import org.onlab.onos.cluster.LeadershipEvent.Type; | ||
17 | +import org.onlab.onos.cluster.LeadershipEventListener; | ||
18 | +import org.onlab.onos.cluster.LeadershipService; | ||
19 | + | ||
20 | +/** | ||
21 | + * A trivial implementation of the leadership service. | ||
22 | + * <p></p> | ||
23 | + * The service is not distributed, so it can assume there's a single leadership | ||
24 | + * contender. This contender is always granted leadership whenever it asks. | ||
25 | + */ | ||
26 | +@Component(immediate = true) | ||
27 | +@Service | ||
28 | +public class SimpleLeadershipManager implements LeadershipService { | ||
29 | + | ||
30 | + private Set<LeadershipEventListener> listeners = new CopyOnWriteArraySet<>(); | ||
31 | + | ||
32 | + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
33 | + private ClusterService clusterService; | ||
34 | + | ||
35 | + private Map<String, Boolean> elections = new ConcurrentHashMap<>(); | ||
36 | + | ||
37 | + @Override | ||
38 | + public ControllerNode getLeader(String path) { | ||
39 | + return elections.get(path) ? clusterService.getLocalNode() : null; | ||
40 | + } | ||
41 | + | ||
42 | + @Override | ||
43 | + public void runForLeadership(String path) { | ||
44 | + elections.put(path, true); | ||
45 | + for (LeadershipEventListener listener : listeners) { | ||
46 | + listener.event(new LeadershipEvent(Type.LEADER_ELECTED, | ||
47 | + new Leadership(path, clusterService.getLocalNode(), 0))); | ||
48 | + } | ||
49 | + } | ||
50 | + | ||
51 | + @Override | ||
52 | + public void withdraw(String path) { | ||
53 | + elections.remove(path); | ||
54 | + for (LeadershipEventListener listener : listeners) { | ||
55 | + listener.event(new LeadershipEvent(Type.LEADER_BOOTED, | ||
56 | + new Leadership(path, clusterService.getLocalNode(), 0))); | ||
57 | + } | ||
58 | + } | ||
59 | + | ||
60 | + @Override | ||
61 | + public Map<String, Leadership> getLeaderBoard() { | ||
62 | + throw new UnsupportedOperationException("I don't know what to do." + | ||
63 | + " I wish you luck."); | ||
64 | + } | ||
65 | + | ||
66 | + @Override | ||
67 | + public void addListener(LeadershipEventListener listener) { | ||
68 | + listeners.add(listener); | ||
69 | + } | ||
70 | + | ||
71 | + @Override | ||
72 | + public void removeListener(LeadershipEventListener listener) { | ||
73 | + listeners.remove(listener); | ||
74 | + } | ||
75 | + | ||
76 | +} |
-
Please register or login to post a comment