2 * Copyright (c) 2015 Huawei Technologies Co., Ltd. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.common.actor;
11 import akka.actor.Address;
12 import akka.actor.Props;
13 import akka.actor.UntypedAbstractActor;
14 import akka.cluster.Cluster;
15 import akka.cluster.ClusterEvent;
16 import akka.japi.Effect;
17 import akka.remote.AssociationErrorEvent;
18 import akka.remote.RemotingLifecycleEvent;
19 import akka.remote.artery.ThisActorSystemQuarantinedEvent;
20 import java.util.HashSet;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
26 * This class listens to Akka RemotingLifecycleEvent events to detect when this node has been
27 * quarantined by another. Once this node gets quarantined, restart the ActorSystem to allow this
28 * node to rejoin the cluster.
30 * @author Gary Wu gary.wu1@huawei.com
33 public class QuarantinedMonitorActor extends UntypedAbstractActor {
34 public static final String ADDRESS = "quarantined-monitor";
36 private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
37 private static final Integer MESSAGE_THRESHOLD = 10;
39 private final Effect callback;
40 private boolean quarantined;
42 private final Set<Address> addressSet = new HashSet<>();
43 private int count = 0;
45 protected QuarantinedMonitorActor(final Effect callback) {
46 this.callback = callback;
48 LOG.debug("Created QuarantinedMonitorActor");
50 getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
51 getContext().system().eventStream().subscribe(getSelf(), ClusterEvent.MemberDowned.class);
55 public void postStop() {
56 LOG.debug("Stopping QuarantinedMonitorActor");
60 public void onReceive(final Object message) throws Exception {
61 final String messageType = message.getClass().getSimpleName();
62 LOG.trace("onReceive {} {}", messageType, message);
64 // check to see if we got quarantined by another node
69 if (message instanceof ThisActorSystemQuarantinedEvent) {
70 final ThisActorSystemQuarantinedEvent event = (ThisActorSystemQuarantinedEvent) message;
71 LOG.warn("Got quarantined by {}", event.remoteAddress());
74 // execute the callback
76 } else if (message instanceof AssociationErrorEvent) {
77 final String errorMessage = message.toString();
78 LOG.trace("errorMessage:{}", errorMessage);
79 if (errorMessage.contains("The remote system has a UID that has been quarantined")) {
80 final Address address = ((AssociationErrorEvent) message).getRemoteAddress();
81 addressSet.add(address);
83 LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
84 if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
87 final AssociationErrorEvent event = (AssociationErrorEvent) message;
88 LOG.warn("Got quarantined via AssociationEvent by {}", event.remoteAddress());
91 // execute the callback
94 } else if (errorMessage.contains("The remote system explicitly disassociated")) {
98 } else if (message instanceof ClusterEvent.MemberDowned) {
99 final ClusterEvent.MemberDowned event = (ClusterEvent.MemberDowned) message;
100 if (Cluster.get(getContext().system()).selfMember().equals(event.member())) {
101 LOG.warn("This member has been downed, restarting");
108 public static Props props(final Effect callback) {
109 return Props.create(QuarantinedMonitorActor.class, callback);