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
  81     public NetconfTopologyManager(final DataBroker dataBroker, final RpcProviderRegistry rpcProviderRegistry,
  82                                   final ClusterSingletonServiceProvider clusterSingletonServiceProvider,
  83                                   final ScheduledThreadPool keepaliveExecutor, final ThreadPool processingExecutor,
  84                                   final ActorSystemProvider actorSystemProvider, final EventExecutor eventExecutor,
  85                                   final NetconfClientDispatcher clientDispatcher, final String topologyId,
  86                                   final Config config, final DOMMountPointService mountPointService) {
  87         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  88         this.rpcProviderRegistry = Preconditions.checkNotNull(rpcProviderRegistry);
  89         this.clusterSingletonServiceProvider = Preconditions.checkNotNull(clusterSingletonServiceProvider);
  90         this.keepaliveExecutor = Preconditions.checkNotNull(keepaliveExecutor);
  91         this.processingExecutor = Preconditions.checkNotNull(processingExecutor);
  92         this.actorSystem = Preconditions.checkNotNull(actorSystemProvider).getActorSystem();
  93         this.eventExecutor = Preconditions.checkNotNull(eventExecutor);
  94         this.clientDispatcher = Preconditions.checkNotNull(clientDispatcher);
  95         this.topologyId = Preconditions.checkNotNull(topologyId);
  96         this.writeTxIdleTimeout = Duration.apply(config.getWriteTransactionIdleTimeout(), TimeUnit.SECONDS);
  97         this.mountPointService = mountPointService;
  98     }
  99
 100     // Blueprint init method
 101     public void init() {
 102         dataChangeListenerRegistration = registerDataTreeChangeListener(topologyId);
 103     }
 104
 105     @Override
 106     public void onDataTreeChanged(@Nonnull final Collection<DataTreeModification<Node>> changes) {
 107         for (final DataTreeModification<Node> change : changes) {
 108             final DataObjectModification<Node> rootNode = change.getRootNode();
 109             final InstanceIdentifier<Node> dataModifIdent = change.getRootPath().getRootIdentifier();
 110             final NodeId nodeId = NetconfTopologyUtils.getNodeId(rootNode.getIdentifier());
 111             switch (rootNode.getModificationType()) {
 112                 case SUBTREE_MODIFIED:
 113                     LOG.debug("Config for node {} updated", nodeId);
 114                     refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 115                     break;
 116                 case WRITE:
 117                     if (contexts.containsKey(dataModifIdent)) {
 118                         LOG.debug("RemoteDevice{{}} was already configured, reconfiguring node...", nodeId);
 119                         refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 120                     } else {
 121                         LOG.debug("Config for node {} created", nodeId);
 122                         startNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 123                     }
 124                     break;
 125                 case DELETE:
 126                     LOG.debug("Config for node {} deleted", nodeId);
 127                     stopNetconfDeviceContext(dataModifIdent);
 128                     break;
 129                 default:
 130                     LOG.warn("Unknown operation for {}.", nodeId);
 131             }
 132         }
 133     }
 134
 135     private void refreshNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 136         final NetconfTopologyContext context = contexts.get(instanceIdentifier);
 137         context.refresh(createSetup(instanceIdentifier, node));
 138     }
 139
 140     private void startNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 141         final NetconfNode netconfNode = node.getAugmentation(NetconfNode.class);
 142         Preconditions.checkNotNull(netconfNode);
 143         Preconditions.checkNotNull(netconfNode.getHost());
 144         Preconditions.checkNotNull(netconfNode.getHost().getIpAddress());
 145
 146         final Timeout actorResponseWaitTime = new Timeout(Duration.create(netconfNode.getActorResponseWaitTime(),
 147                 "seconds"));
 148
 149         final ServiceGroupIdentifier serviceGroupIdent =
 150                 ServiceGroupIdentifier.create(instanceIdentifier.toString());
 151
 152         final NetconfTopologyContext newNetconfTopologyContext =
 153                 new NetconfTopologyContext(createSetup(instanceIdentifier, node), serviceGroupIdent,
 154                         actorResponseWaitTime, mountPointService);
 155
 156         final ClusterSingletonServiceRegistration clusterSingletonServiceRegistration =
 157                 clusterSingletonServiceProvider.registerClusterSingletonService(newNetconfTopologyContext);
 158
 159         clusterRegistrations.put(instanceIdentifier, clusterSingletonServiceRegistration);
 160         contexts.put(instanceIdentifier, newNetconfTopologyContext);
 161     }
 162
 163     private void stopNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier) {
 164         if (contexts.containsKey(instanceIdentifier)) {
 165             try {
 166                 clusterRegistrations.get(instanceIdentifier).close();
 167                 contexts.get(instanceIdentifier).closeFinal();
 168             } catch (final Exception e) {
 169                 LOG.warn("Error at closing topology context. InstanceIdentifier: " + instanceIdentifier);
 170             }
 171             contexts.remove(instanceIdentifier);
 172             clusterRegistrations.remove(instanceIdentifier);
 173         }
 174     }
 175
 176     @Override
 177     public void close() {
 178         if (dataChangeListenerRegistration != null) {
 179             dataChangeListenerRegistration.close();
 180             dataChangeListenerRegistration = null;
 181         }
 182         contexts.forEach((instanceIdentifier, netconfTopologyContext) -> {
 183             try {
 184                 netconfTopologyContext.closeFinal();
 185             } catch (final Exception e) {
 186                 LOG.error("Error at closing topology context. InstanceIdentifier: " + instanceIdentifier, e);
 187             }
 188         });
 189         clusterRegistrations.forEach((instanceIdentifier, clusterSingletonServiceRegistration) -> {
 190             try {
 191                 clusterSingletonServiceRegistration.close();
 192             } catch (final Exception e) {
 193                 LOG.error("Error at unregistering from cluster. InstanceIdentifier: " + instanceIdentifier, e);
 194             }
 195         });
 196         contexts.clear();
 197         clusterRegistrations.clear();
 198     }
 199
 200     private ListenerRegistration<NetconfTopologyManager> registerDataTreeChangeListener(final String topologyId) {
 201         final WriteTransaction wtx = dataBroker.newWriteOnlyTransaction();
 202         initTopology(wtx, LogicalDatastoreType.CONFIGURATION, topologyId);
 203         initTopology(wtx, LogicalDatastoreType.OPERATIONAL, topologyId);
 204         Futures.addCallback(wtx.submit(), new FutureCallback<Void>() {
 205             @Override
 206             public void onSuccess(final Void result) {
 207                 LOG.debug("topology initialization successful");
 208             }
 209
 210             @Override
 211             public void onFailure(@Nonnull final Throwable throwable) {
 212                 LOG.error("Unable to initialize netconf-topology, {}", throwable);
 213             }
 214         });
 215
 216         LOG.debug("Registering datastore listener");
 217         return dataBroker.registerDataTreeChangeListener(
 218                 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION,
 219                         NetconfTopologyUtils.createTopologyListPath(topologyId).child(Node.class)), this);
 220     }
 221
 222     private void initTopology(final WriteTransaction wtx, final LogicalDatastoreType datastoreType, final String topologyId) {
 223         final NetworkTopology networkTopology = new NetworkTopologyBuilder().build();
 224         final InstanceIdentifier<NetworkTopology> networkTopologyId =
 225                 InstanceIdentifier.builder(NetworkTopology.class).build();
 226         wtx.merge(datastoreType, networkTopologyId, networkTopology);
 227         final Topology topology = new TopologyBuilder().setTopologyId(new TopologyId(topologyId)).build();
 228         wtx.merge(datastoreType, networkTopologyId.child(Topology.class,
 229                 new TopologyKey(new TopologyId(topologyId))), topology);
 230     }
 231
 232     private NetconfTopologySetup createSetup(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 233         final NetconfTopologySetupBuilder builder = NetconfTopologySetupBuilder.create()
 234                 .setClusterSingletonServiceProvider(clusterSingletonServiceProvider)
 235                 .setDataBroker(dataBroker)
 236                 .setInstanceIdentifier(instanceIdentifier)
 237                 .setRpcProviderRegistry(rpcProviderRegistry)
 238                 .setNode(node)
 239                 .setActorSystem(actorSystem)
 240                 .setEventExecutor(eventExecutor)
 241                 .setKeepaliveExecutor(keepaliveExecutor)
 242                 .setProcessingExecutor(processingExecutor)
 243                 .setTopologyId(topologyId)
 244                 .setNetconfClientDispatcher(clientDispatcher)
 245                 .setSchemaResourceDTO(NetconfTopologyUtils.setupSchemaCacheDTO(node))
 246                 .setIdleTimeout(writeTxIdleTimeout);
 247
 248         return builder.build();
 249     }
 250 }