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.japi.Effect;
15 import akka.remote.AssociationErrorEvent;
16 import akka.remote.RemotingLifecycleEvent;
17 import akka.remote.ThisActorSystemQuarantinedEvent;
18 import java.util.HashSet;
20 import org.slf4j.Logger;
21 import org.slf4j.LoggerFactory;
24 * This class listens to Akka RemotingLifecycleEvent events to detect when this node has been
25 * quarantined by another. Once this node gets quarantined, restart the ActorSystem to allow this
26 * node to rejoin the cluster.
28 * @author Gary Wu gary.wu1@huawei.com
31 public class QuarantinedMonitorActor extends UntypedAbstractActor {
32 public static final String ADDRESS = "quarantined-monitor";
34 private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
35 private static final Integer MESSAGE_THRESHOLD = 10;
37 private final Effect callback;
38 private boolean quarantined;
40 private Set<Address> addressSet = new HashSet<>();
41 private int count = 0;
43 protected QuarantinedMonitorActor(final Effect callback) {
44 this.callback = callback;
46 LOG.debug("Created QuarantinedMonitorActor");
48 getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
52 public void postStop() {
53 LOG.debug("Stopping QuarantinedMonitorActor");
57 public void onReceive(final Object message) throws Exception {
58 final String messageType = message.getClass().getSimpleName();
59 LOG.trace("onReceive {} {}", messageType, message);
61 // check to see if we got quarantined by another node
66 if (message instanceof ThisActorSystemQuarantinedEvent) {
67 final ThisActorSystemQuarantinedEvent event = (ThisActorSystemQuarantinedEvent) message;
68 LOG.warn("Got quarantined by {}", event.remoteAddress());
71 // execute the callback
73 } else if (message instanceof AssociationErrorEvent) {
74 String errorMessage = message.toString();
75 LOG.trace("errorMessage:{}", errorMessage);
76 if (errorMessage.contains("The remote system has a UID that has been quarantined")) {
77 Address address = ((AssociationErrorEvent) message).getRemoteAddress();
78 addressSet.add(address);
80 LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
81 if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
84 final AssociationErrorEvent event = (AssociationErrorEvent) message;
85 LOG.warn("Got quarantined via AssociationEvent by {}", event.remoteAddress());
88 // execute the callback
91 } else if (errorMessage.contains("The remote system explicitly disassociated")) {
98 public static Props props(final Effect callback) {
99 return Props.create(QuarantinedMonitorActor.class, callback);