blob: 9ceffeed99cc3c358d35dde533ab860fcb8ba622 [file] [log] [blame]
Madan Jampani620f70d2016-01-30 22:22:47 -08001/*
2 * Copyright 2016 Open Networking Laboratory
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16package org.onosproject.cluster.impl;
17
18import static org.slf4j.LoggerFactory.getLogger;
19
20import java.util.Map;
21import java.util.concurrent.Executors;
22import java.util.concurrent.ScheduledExecutorService;
23import java.util.concurrent.TimeUnit;
24
25import org.apache.felix.scr.annotations.Activate;
26import org.apache.felix.scr.annotations.Component;
27import org.apache.felix.scr.annotations.Deactivate;
28import org.apache.felix.scr.annotations.Reference;
29import org.apache.felix.scr.annotations.ReferenceCardinality;
30import org.apache.felix.scr.annotations.Service;
31import org.onlab.util.Tools;
32import org.onosproject.cluster.ClusterService;
33import org.onosproject.cluster.ControllerNode;
34import org.onosproject.cluster.Leadership;
35import org.onosproject.cluster.LeadershipAdminService;
36import org.onosproject.cluster.LeadershipEvent;
37import org.onosproject.cluster.LeadershipEventListener;
38import org.onosproject.cluster.LeadershipService;
39import org.onosproject.cluster.LeadershipStore;
40import org.onosproject.cluster.LeadershipStoreDelegate;
41import org.onosproject.cluster.NodeId;
42import org.onosproject.event.AbstractListenerManager;
43import org.slf4j.Logger;
44
45import com.google.common.collect.Maps;
46
47/**
48 * Implementation of {@link LeadershipService} and {@link LeadershipAdminService}.
49 */
50@Component(immediate = true)
51@Service
52public class LeadershipManager
53 extends AbstractListenerManager<LeadershipEvent, LeadershipEventListener>
54 implements LeadershipService, LeadershipAdminService {
55
56 private final Logger log = getLogger(getClass());
57
58 private LeadershipStoreDelegate delegate = this::post;
59
60 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
61 protected ClusterService clusterService;
62
63 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
64 protected LeadershipStore store;
65
66 private NodeId localNodeId;
67
68 private final ScheduledExecutorService deadlockDetector =
69 Executors.newSingleThreadScheduledExecutor(Tools.groupedThreads("onos/leadership", ""));
70
71 @Activate
72 public void activate() {
73 localNodeId = clusterService.getLocalNode().id();
74 store.setDelegate(delegate);
75 eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry);
76 deadlockDetector.scheduleWithFixedDelay(() -> clusterService.getNodes()
77 .stream()
78 .map(ControllerNode::id)
Thomas Vachuska7a8de842016-03-07 20:56:35 -080079 .filter(id -> !clusterService.getState(id).isActive())
Madan Jampani620f70d2016-01-30 22:22:47 -080080 .forEach(this::unregister), 0, 2, TimeUnit.SECONDS);
81 log.info("Started");
82 }
83
84 @Deactivate
85 public void deactivate() {
86 deadlockDetector.shutdown();
87 Maps.filterValues(store.getLeaderships(), v -> v.candidates().contains(localNodeId))
88 .keySet()
89 .forEach(this::withdraw);
90 store.unsetDelegate(delegate);
91 eventDispatcher.removeSink(LeadershipEvent.class);
92 log.info("Stopped");
93 }
94
95 @Override
96 public Leadership getLeadership(String topic) {
97 return store.getLeadership(topic);
98 }
99
100 @Override
101 public Leadership runForLeadership(String topic) {
102 return store.addRegistration(topic);
103 }
104
105 @Override
106 public void withdraw(String topic) {
107 store.removeRegistration(topic);
108 }
109
110 @Override
111 public Map<String, Leadership> getLeaderBoard() {
112 return store.getLeaderships();
113 }
114
115 @Override
116 public boolean transferLeadership(String topic, NodeId to) {
117 return store.moveLeadership(topic, to);
118 }
119
120 @Override
121 public void unregister(NodeId nodeId) {
122 store.removeRegistration(nodeId);
123 }
124
125 @Override
126 public boolean promoteToTopOfCandidateList(String topic, NodeId nodeId) {
127 return store.makeTopCandidate(topic, nodeId);
128 }
129}