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