opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/common/actor/QuarantinedMonitorActor.java

   1 /*
   2  * Copyright (c) 2015 Huawei Technologies Co., Ltd. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.controller.cluster.common.actor;
  10
  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;
  22 import java.util.Set;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 /**
  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.
  30  *
  31  * @author Gary Wu gary.wu1@huawei.com
  32  *
  33  */
  34 public class QuarantinedMonitorActor extends UntypedAbstractActor {
  35     public static final String ADDRESS = "quarantined-monitor";
  36
  37     private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
  38     private static final Integer MESSAGE_THRESHOLD = 10;
  39
  40     private final Effect callback;
  41     private boolean quarantined;
  42
  43     private final Set<Address> addressSet = new HashSet<>();
  44     private int count = 0;
  45
  46     @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR", justification = "Akka class design")
  47     protected QuarantinedMonitorActor(final Effect callback) {
  48         this.callback = callback;
  49
  50         LOG.debug("Created QuarantinedMonitorActor");
  51
  52         getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
  53         getContext().system().eventStream().subscribe(getSelf(), ClusterEvent.MemberDowned.class);
  54     }
  55
  56     @Override
  57     public void postStop() {
  58         LOG.debug("Stopping QuarantinedMonitorActor");
  59     }
  60
  61     @Override
  62     public void onReceive(final Object message) throws Exception {
  63         final String messageType = message.getClass().getSimpleName();
  64         LOG.trace("onReceive {} {}", messageType, message);
  65
  66         // check to see if we got quarantined by another node
  67         if (quarantined) {
  68             return;
  69         }
  70
  71         if (message instanceof ThisActorSystemQuarantinedEvent) {
  72             final ThisActorSystemQuarantinedEvent event = (ThisActorSystemQuarantinedEvent) message;
  73             LOG.warn("Got quarantined by {}", event.remoteAddress());
  74             quarantined = true;
  75
  76             // execute the callback
  77             callback.apply();
  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);
  84                 count++;
  85                 LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
  86                 if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
  87                     count = 0;
  88                     addressSet.clear();
  89                     final AssociationErrorEvent event = (AssociationErrorEvent) message;
  90                     LOG.warn("Got quarantined via AssociationEvent by {}", event.remoteAddress());
  91                     quarantined = true;
  92
  93                     // execute the callback
  94                     callback.apply();
  95                 }
  96             } else if (errorMessage.contains("The remote system explicitly disassociated")) {
  97                 count = 0;
  98                 addressSet.clear();
  99             }
 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");
 104
 105                 callback.apply();
 106             }
 107         }
 108     }
 109
 110     public static Props props(final Effect callback) {
 111         return Props.create(QuarantinedMonitorActor.class, callback);
 112     }
 113 }