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