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.Props;
  12 import akka.actor.UntypedActor;
  13 import akka.japi.Effect;
  14 import akka.remote.AssociationErrorEvent;
  15 import akka.remote.InvalidAssociation;
  16 import akka.remote.RemotingLifecycleEvent;
  17 import org.slf4j.Logger;
  18 import org.slf4j.LoggerFactory;
  19
  20 /**
  21  * This class listens to Akka RemotingLifecycleEvent events to detect when this node has been
  22  * quarantined by another. Once this node gets quarantined, restart the ActorSystem to allow this
  23  * node to rejoin the cluster.
  24  *
  25  * @author Gary Wu gary.wu1@huawei.com
  26  *
  27  */
  28 public class QuarantinedMonitorActor extends UntypedActor {
  29
  30     private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
  31
  32     public static final String ADDRESS = "quarantined-monitor";
  33
  34     private final Effect callback;
  35     private boolean quarantined;
  36
  37     protected QuarantinedMonitorActor(Effect callback) {
  38         this.callback = callback;
  39
  40         LOG.debug("Created QuarantinedMonitorActor");
  41
  42         getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
  43     }
  44
  45     @Override
  46     public void postStop() {
  47         LOG.debug("Stopping QuarantinedMonitorActor");
  48     }
  49
  50     @Override
  51     public void onReceive(Object message) throws Exception {
  52         final String messageType = message.getClass().getSimpleName();
  53         LOG.trace("onReceive {} {}", messageType, message);
  54
  55         // check to see if we got quarantined by another node
  56
  57         if (quarantined) {
  58             return;
  59         }
  60
  61         // TODO: follow https://github.com/akka/akka/issues/18758 to see if Akka adds a named
  62         // exception for quarantine detection
  63         if (message instanceof AssociationErrorEvent) {
  64             AssociationErrorEvent event = (AssociationErrorEvent) message;
  65             Throwable cause = event.getCause();
  66             if (cause instanceof InvalidAssociation) {
  67                 Throwable cause2 = ((InvalidAssociation) cause).getCause();
  68                 if (cause2.getMessage().contains("quarantined this system")) {
  69                     quarantined = true;
  70
  71                     LOG.warn("Got quarantined by {}", event.getRemoteAddress());
  72
  73                     // execute the callback
  74                     callback.apply();
  75                 } else {
  76                     LOG.debug("received AssociationErrorEvent, cause: InvalidAssociation", cause2);
  77                 }
  78             } else {
  79                 LOG.debug("received AssociationErrorEvent", cause);
  80             }
  81         }
  82     }
  83
  84     public static Props props(final Effect callback) {
  85         return Props.create(QuarantinedMonitorActor.class, callback);
  86     }
  87 }