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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
21 import java.util.HashSet;
23 import org.slf4j.Logger;
24 import org.slf4j.LoggerFactory;
27 * This class listens to Akka RemotingLifecycleEvent events to detect when this node has been
28 * quarantined by another. Once this node gets quarantined, restart the ActorSystem to allow this
29 * node to rejoin the cluster.
31 * @author Gary Wu gary.wu1@huawei.com
34 public class QuarantinedMonitorActor extends UntypedAbstractActor {
35 public static final String ADDRESS = "quarantined-monitor";
37 private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
38 private static final Integer MESSAGE_THRESHOLD = 10;
40 private final Effect callback;
41 private boolean quarantined;
43 private final Set<Address> addressSet = new HashSet<>();
44 private int count = 0;
46 @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR", justification = "Akka class design")
47 protected QuarantinedMonitorActor(final Effect callback) {
48 this.callback = callback;
50 LOG.debug("Created QuarantinedMonitorActor");
52 getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
53 getContext().system().eventStream().subscribe(getSelf(), ClusterEvent.MemberDowned.class);
57 public void postStop() {
58 LOG.debug("Stopping QuarantinedMonitorActor");
62 public void onReceive(final Object message) throws Exception {
63 final String messageType = message.getClass().getSimpleName();
64 LOG.trace("onReceive {} {}", messageType, message);
66 // check to see if we got quarantined by another node
71 if (message instanceof ThisActorSystemQuarantinedEvent) {
72 final ThisActorSystemQuarantinedEvent event = (ThisActorSystemQuarantinedEvent) message;
73 LOG.warn("Got quarantined by {}", event.remoteAddress());
76 // execute the callback
78 } else if (message instanceof AssociationErrorEvent) {
79 final String errorMessage = message.toString();
80 LOG.trace("errorMessage:{}", errorMessage);
81 if (errorMessage.contains("The remote system has a UID that has been quarantined")) {
82 final Address address = ((AssociationErrorEvent) message).getRemoteAddress();
83 addressSet.add(address);
85 LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
86 if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
89 final AssociationErrorEvent event = (AssociationErrorEvent) message;
90 LOG.warn("Got quarantined via AssociationEvent by {}", event.remoteAddress());
93 // execute the callback
96 } else if (errorMessage.contains("The remote system explicitly disassociated")) {
100 } else if (message instanceof ClusterEvent.MemberDowned) {
101 final ClusterEvent.MemberDowned event = (ClusterEvent.MemberDowned) message;
102 if (Cluster.get(getContext().system()).selfMember().equals(event.member())) {
103 LOG.warn("This member has been downed, restarting");
110 public static Props props(final Effect callback) {
111 return Props.create(QuarantinedMonitorActor.class, callback);