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