netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfNodeManager.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. 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 package org.opendaylight.netconf.topology.singleton.impl;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.ActorSelection;
  12 import akka.actor.PoisonPill;
  13 import akka.dispatch.OnComplete;
  14 import akka.pattern.AskTimeoutException;
  15 import akka.pattern.Patterns;
  16 import akka.util.Timeout;
  17 import java.util.Collection;
  18 import javax.annotation.Nonnull;
  19 import javax.annotation.concurrent.GuardedBy;
  20 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  21 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  22 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  23 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  24 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  25 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  26 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  27 import org.opendaylight.netconf.topology.singleton.api.NetconfTopologySingletonService;
  28 import org.opendaylight.netconf.topology.singleton.impl.actors.NetconfNodeActor;
  29 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  30 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
  31 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
  32 import org.opendaylight.netconf.topology.singleton.messages.RefreshSlaveActor;
  33 import org.opendaylight.netconf.topology.singleton.messages.UnregisterSlaveMountPoint;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus;
  36 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  37 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  38 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  39 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  40 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43 import scala.concurrent.Future;
  44
  45 /**
  46  * Managing and reacting on data tree changes in specific netconf node when master writes status to the operational
  47  * data store (e.g. handling lifecycle of slave mount point).
  48  */
  49 class NetconfNodeManager
  50         implements ClusteredDataTreeChangeListener<Node>, NetconfTopologySingletonService, AutoCloseable {
  51
  52     private static final Logger LOG = LoggerFactory.getLogger(NetconfNodeManager.class);
  53
  54     private final Timeout actorResponseWaitTime;
  55     private final DOMMountPointService mountPointService;
  56
  57     private volatile NetconfTopologySetup setup;
  58     private volatile ListenerRegistration<NetconfNodeManager> dataChangeListenerRegistration;
  59     private volatile RemoteDeviceId id;
  60
  61     @GuardedBy("this")
  62     private ActorRef slaveActorRef;
  63
  64     @GuardedBy("this")
  65     private boolean closed;
  66
  67     @GuardedBy("this")
  68     private int lastUpdateCount;
  69
  70     NetconfNodeManager(final NetconfTopologySetup setup,
  71                        final RemoteDeviceId id, final Timeout actorResponseWaitTime,
  72                        final DOMMountPointService mountPointService) {
  73         this.setup = setup;
  74         this.id = id;
  75         this.actorResponseWaitTime = actorResponseWaitTime;
  76         this.mountPointService = mountPointService;
  77     }
  78
  79     @Override
  80     public void onDataTreeChanged(@Nonnull final Collection<DataTreeModification<Node>> changes) {
  81         for (final DataTreeModification<Node> change : changes) {
  82             final DataObjectModification<Node> rootNode = change.getRootNode();
  83             final NodeId nodeId = NetconfTopologyUtils.getNodeId(rootNode.getIdentifier());
  84             switch (rootNode.getModificationType()) {
  85                 case SUBTREE_MODIFIED:
  86                     LOG.debug("{}: Operational state for node {} - subtree modified from {} to {}",
  87                             id, nodeId, rootNode.getDataBefore(), rootNode.getDataAfter());
  88                     handleSlaveMountPoint(rootNode);
  89                     break;
  90                 case WRITE:
  91                     if (rootNode.getDataBefore() != null) {
  92                         LOG.debug("{}: Operational state for node {} updated from {} to {}",
  93                                 id, nodeId, rootNode.getDataBefore(), rootNode.getDataAfter());
  94                     } else {
  95                         LOG.debug("{}: Operational state for node {} created: {}",
  96                                 id, nodeId, rootNode.getDataAfter());
  97                     }
  98                     handleSlaveMountPoint(rootNode);
  99                     break;
 100                 case DELETE:
 101                     LOG.debug("{}: Operational state for node {} deleted.", id, nodeId);
 102                     unregisterSlaveMountpoint();
 103                     break;
 104                 default:
 105                     LOG.debug("{}: Uknown operation for node: {}", id, nodeId);
 106             }
 107         }
 108     }
 109
 110     @Override
 111     public synchronized void close() {
 112         if (closed) {
 113             return;
 114         }
 115
 116         closed = true;
 117         closeActor();
 118         if (dataChangeListenerRegistration != null) {
 119             dataChangeListenerRegistration.close();
 120             dataChangeListenerRegistration = null;
 121         }
 122     }
 123
 124     @GuardedBy("this")
 125     private void closeActor() {
 126         if (slaveActorRef != null) {
 127             LOG.debug("{}: Sending poison pill to {}", id, slaveActorRef);
 128             slaveActorRef.tell(PoisonPill.getInstance(), ActorRef.noSender());
 129             slaveActorRef = null;
 130         }
 131     }
 132
 133     private synchronized void unregisterSlaveMountpoint() {
 134         lastUpdateCount++;
 135         if (slaveActorRef != null) {
 136             LOG.debug("{}: Sending message to unregister slave mountpoint to {}", id, slaveActorRef);
 137             slaveActorRef.tell(new UnregisterSlaveMountPoint(), ActorRef.noSender());
 138         }
 139     }
 140
 141     void registerDataTreeChangeListener(final String topologyId, final NodeKey key) {
 142         final InstanceIdentifier<Node> path = NetconfTopologyUtils.createTopologyNodeListPath(key, topologyId);
 143         LOG.debug("{}: Registering data tree change listener on path {}", id, path);
 144         dataChangeListenerRegistration = setup.getDataBroker().registerDataTreeChangeListener(
 145                 DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL, path), this);
 146     }
 147
 148     private synchronized void handleSlaveMountPoint(final DataObjectModification<Node> rootNode) {
 149         if (closed) {
 150             return;
 151         }
 152
 153         @SuppressWarnings("ConstantConditions")
 154         final NetconfNode netconfNodeAfter = rootNode.getDataAfter().augmentation(NetconfNode.class);
 155
 156         if (NetconfNodeConnectionStatus.ConnectionStatus.Connected.equals(netconfNodeAfter.getConnectionStatus())) {
 157             lastUpdateCount++;
 158             createOrUpdateActorRef();
 159
 160             final String masterAddress = netconfNodeAfter.getClusteredConnectionStatus().getNetconfMasterNode();
 161             final String masterActorPath = NetconfTopologyUtils.createActorPath(masterAddress,
 162                     NetconfTopologyUtils.createMasterActorName(id.getName(),
 163                             netconfNodeAfter.getClusteredConnectionStatus().getNetconfMasterNode()));
 164
 165             final AskForMasterMountPoint askForMasterMountPoint = new AskForMasterMountPoint(slaveActorRef);
 166             final ActorSelection masterActor = setup.getActorSystem().actorSelection(masterActorPath);
 167
 168             LOG.debug("{}: Sending {} message to master {}", id, askForMasterMountPoint, masterActor);
 169
 170             sendAskForMasterMountPointWithRetries(askForMasterMountPoint, masterActor, 1, lastUpdateCount);
 171         } else {
 172             unregisterSlaveMountpoint();
 173         }
 174     }
 175
 176     @GuardedBy("this")
 177     private void sendAskForMasterMountPointWithRetries(final AskForMasterMountPoint askForMasterMountPoint,
 178             final ActorSelection masterActor, final int tries, final int updateCount) {
 179         final Future<Object> future = Patterns.ask(masterActor, askForMasterMountPoint, actorResponseWaitTime);
 180         future.onComplete(new OnComplete<Object>() {
 181             @Override
 182             public void onComplete(final Throwable failure, final Object response) {
 183                 synchronized (this) {
 184                     // Ignore the response if we were since closed or another notification update occurred.
 185                     if (closed || updateCount != lastUpdateCount) {
 186                         return;
 187                     }
 188
 189                     if (failure instanceof AskTimeoutException) {
 190                         if (tries <= 5 || tries % 10 == 0) {
 191                             LOG.warn("{}: Failed to send message to {} - retrying...", id, masterActor, failure);
 192                         }
 193                         sendAskForMasterMountPointWithRetries(askForMasterMountPoint, masterActor, tries + 1,
 194                                 updateCount);
 195                     } else if (failure != null) {
 196                         LOG.error("{}: Failed to send message {} to {}. Slave mount point could not be created",
 197                                 id, askForMasterMountPoint, masterActor, failure);
 198                     } else {
 199                         LOG.debug("{}: {} message to {} succeeded", id, askForMasterMountPoint, masterActor);
 200                     }
 201                 }
 202             }
 203         }, setup.getActorSystem().dispatcher());
 204     }
 205
 206     @GuardedBy("this")
 207     private void createOrUpdateActorRef() {
 208         if (slaveActorRef == null) {
 209             slaveActorRef = setup.getActorSystem().actorOf(NetconfNodeActor.props(setup, id, actorResponseWaitTime,
 210                     mountPointService));
 211             LOG.debug("{}: Slave actor created with name {}", id, slaveActorRef);
 212         } else {
 213             slaveActorRef
 214                     .tell(new RefreshSlaveActor(setup, id, actorResponseWaitTime), ActorRef.noSender());
 215         }
 216     }
 217
 218     void refreshDevice(final NetconfTopologySetup netconfTopologyDeviceSetup, final RemoteDeviceId remoteDeviceId) {
 219         setup = netconfTopologyDeviceSetup;
 220         id = remoteDeviceId;
 221     }
 222 }