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 event) {
  72             LOG.warn("Got quarantined by {}", event.remoteAddress());
  73             quarantined = true;
  74
  75             // execute the callback
  76             callback.apply();
  77         } else if (message instanceof AssociationErrorEvent event) {
  78             final String errorMessage = message.toString();
  79             LOG.trace("errorMessage:{}", errorMessage);
  80             if (errorMessage.contains("The remote system has a UID that has been quarantined")) {
  81                 final Address address = event.getRemoteAddress();
  82                 addressSet.add(address);
  83                 count++;
  84                 LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
  85                 if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
  86                     count = 0;
  87                     addressSet.clear();
  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 event) {
  99             if (Cluster.get(getContext().system()).selfMember().equals(event.member())) {
 100                 LOG.warn("This member has been downed, restarting");
 101
 102                 callback.apply();
 103             }
 104         }
 105     }
 106
 107     public static Props props(final Effect callback) {
 108         return Props.create(QuarantinedMonitorActor.class, callback);
 109     }
 110 }