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