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