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