netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfTopologyManager.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.ActorSystem;
  12 import akka.util.Timeout;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import io.netty.util.concurrent.EventExecutor;
  17 import java.util.Collection;
  18 import java.util.HashMap;
  19 import java.util.Map;
  20 import java.util.concurrent.TimeUnit;
  21 import javax.annotation.Nonnull;
  22 import org.opendaylight.controller.cluster.ActorSystemProvider;
  23 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  24 import org.opendaylight.controller.config.threadpool.ThreadPool;
  25 import org.opendaylight.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
  26 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  27 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification;
  28 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
  29 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  30 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  31 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  32 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
  33 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  34 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  35 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  36 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  37 import org.opendaylight.netconf.client.NetconfClientDispatcher;
  38 import org.opendaylight.netconf.topology.singleton.api.NetconfTopologySingletonService;
  39 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  40 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup.NetconfTopologySetupBuilder;
  41 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.topology.singleton.config.rev170419.Config;
  44 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  45 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopologyBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  47 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  48 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  49 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  51 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  52 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  53 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  54 import org.slf4j.Logger;
  55 import org.slf4j.LoggerFactory;
  56 import scala.concurrent.duration.Duration;
  57
  58 public class NetconfTopologyManager
  59         implements ClusteredDataTreeChangeListener<Node>, NetconfTopologySingletonService, AutoCloseable {
  60
  61     private static final Logger LOG = LoggerFactory.getLogger(NetconfTopologyManager.class);
  62
  63     private final Map<InstanceIdentifier<Node>, NetconfTopologyContext> contexts = new HashMap<>();
  64     private final Map<InstanceIdentifier<Node>, ClusterSingletonServiceRegistration>
  65             clusterRegistrations = new HashMap<>();
  66
  67     private final DataBroker dataBroker;
  68     private final RpcProviderRegistry rpcProviderRegistry;
  69     private final ClusterSingletonServiceProvider clusterSingletonServiceProvider;
  70     private final ScheduledThreadPool keepaliveExecutor;
  71     private final ThreadPool processingExecutor;
  72     private final ActorSystem actorSystem;
  73     private final EventExecutor eventExecutor;
  74     private final NetconfClientDispatcher clientDispatcher;
  75     private final String topologyId;
  76     private final Duration writeTxIdleTimeout;
  77     private final DOMMountPointService mountPointService;
  78
  79     private ListenerRegistration<NetconfTopologyManager> dataChangeListenerRegistration;
  80     private String privateKeyPath;
  81     private String privateKeyPassphrase;
  82
  83     public NetconfTopologyManager(final DataBroker dataBroker, final RpcProviderRegistry rpcProviderRegistry,
  84                                   final ClusterSingletonServiceProvider clusterSingletonServiceProvider,
  85                                   final ScheduledThreadPool keepaliveExecutor, final ThreadPool processingExecutor,
  86                                   final ActorSystemProvider actorSystemProvider, final EventExecutor eventExecutor,
  87                                   final NetconfClientDispatcher clientDispatcher, final String topologyId,
  88                                   final Config config, final DOMMountPointService mountPointService) {
  89         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  90         this.rpcProviderRegistry = Preconditions.checkNotNull(rpcProviderRegistry);
  91         this.clusterSingletonServiceProvider = Preconditions.checkNotNull(clusterSingletonServiceProvider);
  92         this.keepaliveExecutor = Preconditions.checkNotNull(keepaliveExecutor);
  93         this.processingExecutor = Preconditions.checkNotNull(processingExecutor);
  94         this.actorSystem = Preconditions.checkNotNull(actorSystemProvider).getActorSystem();
  95         this.eventExecutor = Preconditions.checkNotNull(eventExecutor);
  96         this.clientDispatcher = Preconditions.checkNotNull(clientDispatcher);
  97         this.topologyId = Preconditions.checkNotNull(topologyId);
  98         this.writeTxIdleTimeout = Duration.apply(config.getWriteTransactionIdleTimeout(), TimeUnit.SECONDS);
  99         this.mountPointService = mountPointService;
 100     }
 101
 102     // Blueprint init method
 103     public void init() {
 104         dataChangeListenerRegistration = registerDataTreeChangeListener(topologyId);
 105     }
 106
 107     @Override
 108     public void onDataTreeChanged(@Nonnull final Collection<DataTreeModification<Node>> changes) {
 109         for (final DataTreeModification<Node> change : changes) {
 110             final DataObjectModification<Node> rootNode = change.getRootNode();
 111             final InstanceIdentifier<Node> dataModifIdent = change.getRootPath().getRootIdentifier();
 112             final NodeId nodeId = NetconfTopologyUtils.getNodeId(rootNode.getIdentifier());
 113             switch (rootNode.getModificationType()) {
 114                 case SUBTREE_MODIFIED:
 115                     LOG.debug("Config for node {} updated", nodeId);
 116                     refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 117                     break;
 118                 case WRITE:
 119                     if (contexts.containsKey(dataModifIdent)) {
 120                         LOG.debug("RemoteDevice{{}} was already configured, reconfiguring node...", nodeId);
 121                         refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 122                     } else {
 123                         LOG.debug("Config for node {} created", nodeId);
 124                         startNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 125                     }
 126                     break;
 127                 case DELETE:
 128                     LOG.debug("Config for node {} deleted", nodeId);
 129                     stopNetconfDeviceContext(dataModifIdent);
 130                     break;
 131                 default:
 132                     LOG.warn("Unknown operation for {}.", nodeId);
 133             }
 134         }
 135     }
 136
 137     private void refreshNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 138         final NetconfTopologyContext context = contexts.get(instanceIdentifier);
 139         context.refresh(createSetup(instanceIdentifier, node));
 140     }
 141
 142     // ClusterSingletonServiceRegistration registerClusterSingletonService method throws a Runtime exception if there
 143     // are problems with registration and client has to deal with it. Only thing we can do if this error occurs is to
 144     // retry registration several times and log the error.
 145     // TODO change to a specific documented Exception when changed in ClusterSingletonServiceProvider
 146     @SuppressWarnings("checkstyle:IllegalCatch")
 147     private void startNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 148         final NetconfNode netconfNode = node.getAugmentation(NetconfNode.class);
 149         Preconditions.checkNotNull(netconfNode);
 150         Preconditions.checkNotNull(netconfNode.getHost());
 151         Preconditions.checkNotNull(netconfNode.getHost().getIpAddress());
 152
 153         final Timeout actorResponseWaitTime = new Timeout(Duration.create(netconfNode.getActorResponseWaitTime(),
 154                 "seconds"));
 155
 156         final ServiceGroupIdentifier serviceGroupIdent =
 157                 ServiceGroupIdentifier.create(instanceIdentifier.toString());
 158
 159         final NetconfTopologyContext newNetconfTopologyContext =
 160                 new NetconfTopologyContext(createSetup(instanceIdentifier, node), serviceGroupIdent,
 161                         actorResponseWaitTime, mountPointService);
 162
 163         int tries = 3;
 164         while (true) {
 165             try {
 166                 final ClusterSingletonServiceRegistration clusterSingletonServiceRegistration =
 167                         clusterSingletonServiceProvider.registerClusterSingletonService(newNetconfTopologyContext);
 168                 clusterRegistrations.put(instanceIdentifier, clusterSingletonServiceRegistration);
 169                 contexts.put(instanceIdentifier, newNetconfTopologyContext);
 170                 break;
 171             } catch (final RuntimeException e) {
 172                 LOG.warn("Unable to register cluster singleton service {}, trying again", newNetconfTopologyContext, e);
 173
 174                 if (--tries <= 0) {
 175                     LOG.error("Unable to register cluster singleton service {} - done trying, closing topology context",
 176                             newNetconfTopologyContext, e);
 177                     close();
 178                     break;
 179                 }
 180             }
 181         }
 182
 183     }
 184
 185     @SuppressWarnings("checkstyle:IllegalCatch")
 186     private void stopNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier) {
 187         if (contexts.containsKey(instanceIdentifier)) {
 188             try {
 189                 clusterRegistrations.get(instanceIdentifier).close();
 190                 contexts.get(instanceIdentifier).closeFinal();
 191             } catch (final Exception e) {
 192                 LOG.warn("Error at closing topology context. InstanceIdentifier: " + instanceIdentifier);
 193             }
 194             contexts.remove(instanceIdentifier);
 195             clusterRegistrations.remove(instanceIdentifier);
 196         }
 197     }
 198
 199     @SuppressWarnings("checkstyle:IllegalCatch")
 200     @Override
 201     public void close() {
 202         if (dataChangeListenerRegistration != null) {
 203             dataChangeListenerRegistration.close();
 204             dataChangeListenerRegistration = null;
 205         }
 206         contexts.forEach((instanceIdentifier, netconfTopologyContext) -> {
 207             try {
 208                 netconfTopologyContext.closeFinal();
 209             } catch (final Exception e) {
 210                 LOG.error("Error at closing topology context. InstanceIdentifier: " + instanceIdentifier, e);
 211             }
 212         });
 213         clusterRegistrations.forEach((instanceIdentifier, clusterSingletonServiceRegistration) -> {
 214             try {
 215                 clusterSingletonServiceRegistration.close();
 216             } catch (final Exception e) {
 217                 LOG.error("Error at unregistering from cluster. InstanceIdentifier: " + instanceIdentifier, e);
 218             }
 219         });
 220         contexts.clear();
 221         clusterRegistrations.clear();
 222     }
 223
 224     /**
 225      * Sets the private key path from location specified in configuration file using blueprint.
 226      */
 227     public void setPrivateKeyPath(String privateKeyPath) {
 228         this.privateKeyPath = privateKeyPath;
 229     }
 230
 231     /**
 232      * Sets the private key passphrase from location specified in configuration file using blueprint.
 233      */
 234     public void setPrivateKeyPassphrase(String privateKeyPassphrase) {
 235         this.privateKeyPassphrase = privateKeyPassphrase;
 236     }
 237
 238     private ListenerRegistration<NetconfTopologyManager> registerDataTreeChangeListener(final String topologyId) {
 239         final WriteTransaction wtx = dataBroker.newWriteOnlyTransaction();
 240         initTopology(wtx, LogicalDatastoreType.CONFIGURATION, topologyId);
 241         initTopology(wtx, LogicalDatastoreType.OPERATIONAL, topologyId);
 242         Futures.addCallback(wtx.submit(), new FutureCallback<Void>() {
 243             @Override
 244             public void onSuccess(final Void result) {
 245                 LOG.debug("topology initialization successful");
 246             }
 247
 248             @Override
 249             public void onFailure(@Nonnull final Throwable throwable) {
 250                 LOG.error("Unable to initialize netconf-topology, {}", throwable);
 251             }
 252         });
 253
 254         LOG.debug("Registering datastore listener");
 255         return dataBroker.registerDataTreeChangeListener(
 256                 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION,
 257                         NetconfTopologyUtils.createTopologyListPath(topologyId).child(Node.class)), this);
 258     }
 259
 260     private void initTopology(final WriteTransaction wtx, final LogicalDatastoreType datastoreType,
 261                               final String topologyId) {
 262         final NetworkTopology networkTopology = new NetworkTopologyBuilder().build();
 263         final InstanceIdentifier<NetworkTopology> networkTopologyId =
 264                 InstanceIdentifier.builder(NetworkTopology.class).build();
 265         wtx.merge(datastoreType, networkTopologyId, networkTopology);
 266         final Topology topology = new TopologyBuilder().setTopologyId(new TopologyId(topologyId)).build();
 267         wtx.merge(datastoreType, networkTopologyId.child(Topology.class,
 268                 new TopologyKey(new TopologyId(topologyId))), topology);
 269     }
 270
 271     private NetconfTopologySetup createSetup(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 272         final NetconfTopologySetupBuilder builder = NetconfTopologySetupBuilder.create()
 273                 .setClusterSingletonServiceProvider(clusterSingletonServiceProvider)
 274                 .setDataBroker(dataBroker)
 275                 .setInstanceIdentifier(instanceIdentifier)
 276                 .setRpcProviderRegistry(rpcProviderRegistry)
 277                 .setNode(node)
 278                 .setActorSystem(actorSystem)
 279                 .setEventExecutor(eventExecutor)
 280                 .setKeepaliveExecutor(keepaliveExecutor)
 281                 .setProcessingExecutor(processingExecutor)
 282                 .setTopologyId(topologyId)
 283                 .setNetconfClientDispatcher(clientDispatcher)
 284                 .setSchemaResourceDTO(NetconfTopologyUtils.setupSchemaCacheDTO(node))
 285                 .setIdleTimeout(writeTxIdleTimeout)
 286                 .setPrivateKeyPath(privateKeyPath)
 287                 .setPrivateKeyPassphrase(privateKeyPassphrase);
 288
 289         return builder.build();
 290     }
 291 }