blob: 5e4e71c78ec9dd674f757a810207b079f40ea158 [file] [log] [blame]
Thomas Vachuska4f1a60c2014-10-28 13:39:07 -07001/*
Brian O'Connor5ab426f2016-04-09 01:19:45 -07002 * Copyright 2014-present Open Networking Laboratory
Thomas Vachuska4f1a60c2014-10-28 13:39:07 -07003 *
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 */
Brian O'Connorabafb502014-12-02 22:26:20 -080016package org.onosproject.store.cluster.impl;
tom2d7c65f2014-09-23 01:09:35 -070017
HIGUCHI Yuta1979f552015-12-28 21:24:26 -080018import com.google.common.base.MoreObjects;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070019import com.google.common.collect.ImmutableSet;
20import com.google.common.collect.Maps;
Madan Jampanie3375062016-04-19 16:32:10 -070021
tom2d7c65f2014-09-23 01:09:35 -070022import org.apache.felix.scr.annotations.Activate;
23import org.apache.felix.scr.annotations.Component;
24import org.apache.felix.scr.annotations.Deactivate;
sangyun-hanf98df542016-03-24 20:28:03 +090025import org.apache.felix.scr.annotations.Modified;
26import org.apache.felix.scr.annotations.Property;
Madan Jampaniafeebbd2015-05-19 15:26:01 -070027import org.apache.felix.scr.annotations.Reference;
28import org.apache.felix.scr.annotations.ReferenceCardinality;
tom2d7c65f2014-09-23 01:09:35 -070029import org.apache.felix.scr.annotations.Service;
Madan Jampani7d2fab22015-03-18 17:21:57 -070030import org.joda.time.DateTime;
Ayaka Koshibedd91b842015-03-02 14:48:47 -080031import org.onlab.packet.IpAddress;
32import org.onlab.util.KryoNamespace;
sangyun-hanf98df542016-03-24 20:28:03 +090033import org.onlab.util.Tools;
Brian O'Connorabafb502014-12-02 22:26:20 -080034import org.onosproject.cluster.ClusterEvent;
Madan Jampaniec1df022015-10-13 21:23:03 -070035import org.onosproject.cluster.ClusterMetadataService;
Brian O'Connorabafb502014-12-02 22:26:20 -080036import org.onosproject.cluster.ClusterStore;
37import org.onosproject.cluster.ClusterStoreDelegate;
38import org.onosproject.cluster.ControllerNode;
Ayaka Koshibedd91b842015-03-02 14:48:47 -080039import org.onosproject.cluster.ControllerNode.State;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070040import org.onosproject.cluster.DefaultControllerNode;
Brian O'Connorabafb502014-12-02 22:26:20 -080041import org.onosproject.cluster.NodeId;
Ayaka Koshibedd91b842015-03-02 14:48:47 -080042import org.onosproject.store.AbstractStore;
Madan Jampanic26eede2015-04-16 11:42:16 -070043import org.onosproject.store.cluster.messaging.Endpoint;
Madan Jampaniafeebbd2015-05-19 15:26:01 -070044import org.onosproject.store.cluster.messaging.MessagingService;
Ayaka Koshibedd91b842015-03-02 14:48:47 -080045import org.onosproject.store.serializers.KryoNamespaces;
HIGUCHI Yutae7290652016-05-18 11:29:01 -070046import org.onosproject.store.serializers.StoreSerializer;
sangyun-hanf98df542016-03-24 20:28:03 +090047import org.osgi.service.component.ComponentContext;
Ayaka Koshibedd91b842015-03-02 14:48:47 -080048import org.slf4j.Logger;
tom2d7c65f2014-09-23 01:09:35 -070049
sangyun-hanf98df542016-03-24 20:28:03 +090050import java.util.Dictionary;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070051import java.util.Map;
Thomas Vachuskafba7f3d2016-03-23 15:46:25 -070052import java.util.Objects;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070053import java.util.Set;
54import java.util.concurrent.ExecutorService;
55import java.util.concurrent.Executors;
56import java.util.concurrent.ScheduledExecutorService;
57import java.util.concurrent.TimeUnit;
Madan Jampanid36def02016-01-13 11:21:56 -080058import java.util.function.BiConsumer;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070059import java.util.stream.Collectors;
60
sangyun-hanf98df542016-03-24 20:28:03 +090061import static com.google.common.base.Preconditions.checkArgument;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070062import static com.google.common.base.Preconditions.checkNotNull;
63import static org.onlab.util.Tools.groupedThreads;
Thomas Vachuskafba7f3d2016-03-23 15:46:25 -070064import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_ACTIVATED;
65import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_DEACTIVATED;
66import static org.onosproject.cluster.ClusterEvent.Type.INSTANCE_READY;
Jonathan Hart4a4d18f2015-03-26 12:16:16 -070067import static org.slf4j.LoggerFactory.getLogger;
tom2d7c65f2014-09-23 01:09:35 -070068
tom2d7c65f2014-09-23 01:09:35 -070069@Component(immediate = true)
70@Service
Ayaka Koshibedd91b842015-03-02 14:48:47 -080071/**
72 * Distributed cluster nodes store that employs an accrual failure
73 * detector to identify cluster member up/down status.
74 */
tom0755a362014-09-24 11:54:43 -070075public class DistributedClusterStore
Ayaka Koshibedd91b842015-03-02 14:48:47 -080076 extends AbstractStore<ClusterEvent, ClusterStoreDelegate>
tomb41d1ac2014-09-24 01:51:24 -070077 implements ClusterStore {
tom2d7c65f2014-09-23 01:09:35 -070078
Thomas Vachuska8dc1a692015-03-31 01:01:37 -070079 private static final Logger log = getLogger(DistributedClusterStore.class);
tom2d7c65f2014-09-23 01:09:35 -070080
Thomas Vachuskade563cf2015-04-01 00:28:50 -070081 public static final String HEARTBEAT_MESSAGE = "onos-cluster-heartbeat";
82
sangyun-hanf98df542016-03-24 20:28:03 +090083 private static final int DEFAULT_HEARTBEAT_INTERVAL = 100;
84 @Property(name = "heartbeatInterval", intValue = DEFAULT_HEARTBEAT_INTERVAL,
85 label = "Interval time to send heartbeat to other controller nodes (millisecond)")
86 private int heartbeatInterval = DEFAULT_HEARTBEAT_INTERVAL;
87
88 private static final int DEFAULT_PHI_FAILURE_THRESHOLD = 10;
89 @Property(name = "phiFailureThreshold", intValue = DEFAULT_PHI_FAILURE_THRESHOLD,
90 label = "the value of Phi threshold to detect accrual failure")
91 private int phiFailureThreshold = DEFAULT_PHI_FAILURE_THRESHOLD;
tom2d7c65f2014-09-23 01:09:35 -070092
HIGUCHI Yutae7290652016-05-18 11:29:01 -070093 private static final StoreSerializer SERIALIZER = StoreSerializer.using(
94 KryoNamespace.newBuilder()
Thomas Vachuska8dc1a692015-03-31 01:01:37 -070095 .register(KryoNamespaces.API)
HIGUCHI Yutae7290652016-05-18 11:29:01 -070096 .nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID)
Thomas Vachuska8dc1a692015-03-31 01:01:37 -070097 .register(HeartbeatMessage.class)
HIGUCHI Yutae7290652016-05-18 11:29:01 -070098 .build("ClusterStore"));
Ayaka Koshibedd91b842015-03-02 14:48:47 -080099
100 private static final String INSTANCE_ID_NULL = "Instance ID cannot be null";
101
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800102 private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap();
103 private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap();
Madan Jampani7d2fab22015-03-18 17:21:57 -0700104 private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap();
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800105
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800106 private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor(
HIGUCHI Yutad9e01052016-04-14 09:31:42 -0700107 groupedThreads("onos/cluster/membership", "heartbeat-sender", log));
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800108 private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor(
HIGUCHI Yutad9e01052016-04-14 09:31:42 -0700109 groupedThreads("onos/cluster/membership", "heartbeat-receiver", log));
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800110
111 private PhiAccrualFailureDetector failureDetector;
112
113 private ControllerNode localNode;
114
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700115 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
Madan Jampaniec1df022015-10-13 21:23:03 -0700116 protected ClusterMetadataService clusterMetadataService;
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700117
118 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
119 protected MessagingService messagingService;
120
tom2d7c65f2014-09-23 01:09:35 -0700121 @Activate
122 public void activate() {
Madan Jampaniec1df022015-10-13 21:23:03 -0700123 localNode = clusterMetadataService.getLocalNode();
Thomas Vachuska8dc1a692015-03-31 01:01:37 -0700124
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800125 messagingService.registerHandler(HEARTBEAT_MESSAGE,
Thomas Vachuska8dc1a692015-03-31 01:01:37 -0700126 new HeartbeatMessageHandler(), heartBeatMessageHandler);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800127
128 failureDetector = new PhiAccrualFailureDetector();
129
130 heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0,
sangyun-hanf98df542016-03-24 20:28:03 +0900131 heartbeatInterval, TimeUnit.MILLISECONDS);
tomb41d1ac2014-09-24 01:51:24 -0700132
133 log.info("Started");
134 }
135
tom2d7c65f2014-09-23 01:09:35 -0700136 @Deactivate
137 public void deactivate() {
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700138 messagingService.unregisterHandler(HEARTBEAT_MESSAGE);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800139 heartBeatSender.shutdownNow();
140 heartBeatMessageHandler.shutdownNow();
141
tom2d7c65f2014-09-23 01:09:35 -0700142 log.info("Stopped");
143 }
144
sangyun-hanf98df542016-03-24 20:28:03 +0900145 @Modified
146 public void modified(ComponentContext context) {
147 readComponentConfiguration(context);
148 restartHeartbeatSender();
149 }
150
tom2d7c65f2014-09-23 01:09:35 -0700151 @Override
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800152 public void setDelegate(ClusterStoreDelegate delegate) {
153 checkNotNull(delegate, "Delegate cannot be null");
154 this.delegate = delegate;
155 }
156
157 @Override
158 public void unsetDelegate(ClusterStoreDelegate delegate) {
159 this.delegate = null;
160 }
161
162 @Override
163 public boolean hasDelegate() {
164 return this.delegate != null;
165 }
166
167 @Override
tom2d7c65f2014-09-23 01:09:35 -0700168 public ControllerNode getLocalNode() {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800169 return localNode;
tom2d7c65f2014-09-23 01:09:35 -0700170 }
171
172 @Override
173 public Set<ControllerNode> getNodes() {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800174 return ImmutableSet.copyOf(allNodes.values());
tom2d7c65f2014-09-23 01:09:35 -0700175 }
176
177 @Override
178 public ControllerNode getNode(NodeId nodeId) {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800179 checkNotNull(nodeId, INSTANCE_ID_NULL);
180 return allNodes.get(nodeId);
tom2d7c65f2014-09-23 01:09:35 -0700181 }
182
183 @Override
tomb41d1ac2014-09-24 01:51:24 -0700184 public State getState(NodeId nodeId) {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800185 checkNotNull(nodeId, INSTANCE_ID_NULL);
HIGUCHI Yuta1979f552015-12-28 21:24:26 -0800186 return MoreObjects.firstNonNull(nodeStates.get(nodeId), State.INACTIVE);
tomb41d1ac2014-09-24 01:51:24 -0700187 }
188
189 @Override
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800190 public void markFullyStarted(boolean started) {
191 updateState(localNode.id(), started ? State.READY : State.ACTIVE);
192 }
193
194 @Override
Pavlin Radoslavov444b5192014-10-28 10:45:19 -0700195 public ControllerNode addNode(NodeId nodeId, IpAddress ip, int tcpPort) {
sangyun-hanf98df542016-03-24 20:28:03 +0900196 checkNotNull(nodeId, INSTANCE_ID_NULL);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800197 ControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort);
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700198 addNode(node);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800199 return node;
tomee49c372014-09-26 15:14:50 -0700200 }
201
202 @Override
tomb41d1ac2014-09-24 01:51:24 -0700203 public void removeNode(NodeId nodeId) {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800204 checkNotNull(nodeId, INSTANCE_ID_NULL);
205 ControllerNode node = allNodes.remove(nodeId);
206 if (node != null) {
207 nodeStates.remove(nodeId);
Jonathan Hartf1141262015-04-23 11:27:07 -0700208 notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_REMOVED, node));
tomb41d1ac2014-09-24 01:51:24 -0700209 }
210 }
211
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700212 private void addNode(ControllerNode node) {
213 allNodes.put(node.id(), node);
Madan Jampaniec1df022015-10-13 21:23:03 -0700214 updateState(node.id(), node.equals(localNode) ? State.ACTIVE : State.INACTIVE);
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700215 notifyDelegate(new ClusterEvent(ClusterEvent.Type.INSTANCE_ADDED, node));
Thomas Vachuskade563cf2015-04-01 00:28:50 -0700216 }
217
Madan Jampani7d2fab22015-03-18 17:21:57 -0700218 private void updateState(NodeId nodeId, State newState) {
Thomas Vachuskafba7f3d2016-03-23 15:46:25 -0700219 State currentState = nodeStates.get(nodeId);
220 if (!Objects.equals(currentState, newState)) {
221 nodeStates.put(nodeId, newState);
222 nodeStateLastUpdatedTimes.put(nodeId, DateTime.now());
223 notifyStateChange(nodeId, currentState, newState);
224 }
Madan Jampani7d2fab22015-03-18 17:21:57 -0700225 }
226
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800227 private void heartbeat() {
228 try {
229 Set<ControllerNode> peers = allNodes.values()
230 .stream()
231 .filter(node -> !(node.id().equals(localNode.id())))
232 .collect(Collectors.toSet());
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800233 State state = nodeStates.get(localNode.id());
Madan Jampanie3375062016-04-19 16:32:10 -0700234 byte[] hbMessagePayload = SERIALIZER.encode(new HeartbeatMessage(localNode, state));
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800235 peers.forEach((node) -> {
236 heartbeatToPeer(hbMessagePayload, node);
237 State currentState = nodeStates.get(node.id());
238 double phi = failureDetector.phi(node.id());
sangyun-hanf98df542016-03-24 20:28:03 +0900239 if (phi >= phiFailureThreshold) {
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800240 if (currentState.isActive()) {
Madan Jampani7d2fab22015-03-18 17:21:57 -0700241 updateState(node.id(), State.INACTIVE);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800242 }
243 } else {
244 if (currentState == State.INACTIVE) {
Madan Jampani7d2fab22015-03-18 17:21:57 -0700245 updateState(node.id(), State.ACTIVE);
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800246 }
247 }
248 });
249 } catch (Exception e) {
250 log.debug("Failed to send heartbeat", e);
251 }
tomb41d1ac2014-09-24 01:51:24 -0700252 }
253
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800254 private void notifyStateChange(NodeId nodeId, State oldState, State newState) {
Thomas Vachuskafba7f3d2016-03-23 15:46:25 -0700255 if (oldState != newState) {
256 ControllerNode node = allNodes.get(nodeId);
Jon Hall1f13d642016-05-13 17:53:01 -0700257 // Either this node or that node is no longer part of the same cluster
258 if (node == null) {
Jon Hall66870baa2016-06-20 12:12:10 -0700259 log.debug("Could not find node {} in the cluster, ignoring state change", nodeId);
Jon Hall1f13d642016-05-13 17:53:01 -0700260 return;
261 }
Thomas Vachuskafba7f3d2016-03-23 15:46:25 -0700262 ClusterEvent.Type type = newState == State.READY ? INSTANCE_READY :
263 newState == State.ACTIVE ? INSTANCE_ACTIVATED :
264 INSTANCE_DEACTIVATED;
265 notifyDelegate(new ClusterEvent(type, node));
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800266 }
tomb41d1ac2014-09-24 01:51:24 -0700267 }
268
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800269 private void heartbeatToPeer(byte[] messagePayload, ControllerNode peer) {
Madan Jampaniafeebbd2015-05-19 15:26:01 -0700270 Endpoint remoteEp = new Endpoint(peer.ip(), peer.tcpPort());
Madan Jampani175e8fd2015-05-20 14:10:45 -0700271 messagingService.sendAsync(remoteEp, HEARTBEAT_MESSAGE, messagePayload).whenComplete((result, error) -> {
272 if (error != null) {
273 log.trace("Sending heartbeat to {} failed", remoteEp, error);
274 }
275 });
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800276 }
277
Madan Jampanid36def02016-01-13 11:21:56 -0800278 private class HeartbeatMessageHandler implements BiConsumer<Endpoint, byte[]> {
tomb41d1ac2014-09-24 01:51:24 -0700279 @Override
Madan Jampanid36def02016-01-13 11:21:56 -0800280 public void accept(Endpoint sender, byte[] message) {
Madan Jampanic26eede2015-04-16 11:42:16 -0700281 HeartbeatMessage hb = SERIALIZER.decode(message);
Madan Jampanie3375062016-04-19 16:32:10 -0700282 if (clusterMetadataService.getClusterMetadata().getNodes().contains(hb.source())) {
283 failureDetector.report(hb.source().id());
284 updateState(hb.source().id(), hb.state);
285 }
tomb41d1ac2014-09-24 01:51:24 -0700286 }
tom2d7c65f2014-09-23 01:09:35 -0700287 }
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800288
289 private static class HeartbeatMessage {
290 private ControllerNode source;
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800291 private State state;
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800292
Madan Jampanie3375062016-04-19 16:32:10 -0700293 public HeartbeatMessage(ControllerNode source, State state) {
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800294 this.source = source;
Thomas Vachuska7a8de842016-03-07 20:56:35 -0800295 this.state = state != null ? state : State.ACTIVE;
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800296 }
297
298 public ControllerNode source() {
299 return source;
300 }
Ayaka Koshibedd91b842015-03-02 14:48:47 -0800301 }
302
Madan Jampani7d2fab22015-03-18 17:21:57 -0700303 @Override
304 public DateTime getLastUpdated(NodeId nodeId) {
305 return nodeStateLastUpdatedTimes.get(nodeId);
306 }
Thomas Vachuskade563cf2015-04-01 00:28:50 -0700307
sangyun-hanf98df542016-03-24 20:28:03 +0900308 /**
309 * Extracts properties from the component configuration context.
310 *
311 * @param context the component context
312 */
313 private void readComponentConfiguration(ComponentContext context) {
314 Dictionary<?, ?> properties = context.getProperties();
315
316 Integer newHeartbeatInterval = Tools.getIntegerProperty(properties,
317 "heartbeatInterval");
318 if (newHeartbeatInterval == null) {
319 setHeartbeatInterval(DEFAULT_HEARTBEAT_INTERVAL);
320 log.info("Heartbeat interval time is not configured, default value is {}",
321 DEFAULT_HEARTBEAT_INTERVAL);
322 } else {
323 setHeartbeatInterval(newHeartbeatInterval);
324 log.info("Configured. Heartbeat interval time is configured to {}",
325 heartbeatInterval);
326 }
327
328 Integer newPhiFailureThreshold = Tools.getIntegerProperty(properties,
329 "phiFailureThreshold");
330 if (newPhiFailureThreshold == null) {
331 setPhiFailureThreshold(DEFAULT_PHI_FAILURE_THRESHOLD);
332 log.info("Phi failure threshold is not configured, default value is {}",
333 DEFAULT_PHI_FAILURE_THRESHOLD);
334 } else {
335 setPhiFailureThreshold(newPhiFailureThreshold);
336 log.info("Configured. Phi failure threshold is configured to {}",
337 phiFailureThreshold);
338 }
339 }
340
341 /**
342 * Sets heartbeat interval between the termination of one execution of heartbeat
343 * and the commencement of the next.
344 *
345 * @param interval term between each heartbeat
346 */
347 private void setHeartbeatInterval(int interval) {
348 try {
349 checkArgument(interval > 0, "Interval must be greater than zero");
350 heartbeatInterval = interval;
351 } catch (IllegalArgumentException e) {
352 log.warn(e.getMessage());
353 heartbeatInterval = DEFAULT_HEARTBEAT_INTERVAL;
354 }
355 }
356
357 /**
358 * Sets Phi failure threshold.
359 * Phi is based on a paper titled: "The φ Accrual Failure Detector" by Hayashibara, et al.
360 *
361 * @param threshold
362 */
363 private void setPhiFailureThreshold(int threshold) {
364 phiFailureThreshold = threshold;
365 }
366
367 /**
368 * Restarts heartbeatSender executor.
sangyun-hanf98df542016-03-24 20:28:03 +0900369 */
370 private void restartHeartbeatSender() {
371 try {
372 ScheduledExecutorService prevSender = heartBeatSender;
373 heartBeatSender = Executors.newSingleThreadScheduledExecutor(
HIGUCHI Yutad9e01052016-04-14 09:31:42 -0700374 groupedThreads("onos/cluster/membership", "heartbeat-sender-%d", log));
sangyun-hanf98df542016-03-24 20:28:03 +0900375 heartBeatSender.scheduleWithFixedDelay(this::heartbeat, 0,
376 heartbeatInterval, TimeUnit.MILLISECONDS);
377 prevSender.shutdown();
378 } catch (Exception e) {
379 log.warn(e.getMessage());
380 }
381 }
Jon Hall1f13d642016-05-13 17:53:01 -0700382}