apps/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 package org.opendaylight.netconf.topology.singleton.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.actor.ActorSystem;
  13 import akka.util.Timeout;
  14 import com.google.common.annotations.VisibleForTesting;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.MoreExecutors;
  17 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  18 import io.netty.util.Timer;
  19 import java.time.Duration;
  20 import java.util.Collection;
  21 import java.util.Map;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import javax.annotation.PreDestroy;
  24 import javax.inject.Inject;
  25 import javax.inject.Singleton;
  26 import org.opendaylight.aaa.encrypt.AAAEncryptionService;
  27 import org.opendaylight.controller.cluster.ActorSystemProvider;
  28 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  29 import org.opendaylight.mdsal.binding.api.DataBroker;
  30 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  31 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  32 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  33 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  34 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  35 import org.opendaylight.mdsal.common.api.CommitInfo;
  36 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  37 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  38 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  39 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  40 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  41 import org.opendaylight.netconf.client.NetconfClientFactory;
  42 import org.opendaylight.netconf.client.mdsal.api.BaseNetconfSchemas;
  43 import org.opendaylight.netconf.client.mdsal.api.DeviceActionFactory;
  44 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
  45 import org.opendaylight.netconf.client.mdsal.api.SchemaResourceManager;
  46 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  47 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
  48 import org.opendaylight.netconf.topology.spi.NetconfClientConfigurationBuilderFactory;
  49 import org.opendaylight.netconf.topology.spi.NetconfNodeUtils;
  50 import org.opendaylight.netconf.topology.spi.NetconfTopologyRPCProvider;
  51 import org.opendaylight.netconf.topology.spi.NetconfTopologySchemaAssembler;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev231121.NetconfNode;
  53 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  54 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  55 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  56 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  57 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  59 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  60 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  61 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  62 import org.opendaylight.yangtools.yang.common.Uint16;
  63 import org.osgi.service.component.annotations.Activate;
  64 import org.osgi.service.component.annotations.Component;
  65 import org.osgi.service.component.annotations.Deactivate;
  66 import org.osgi.service.component.annotations.Reference;
  67 import org.osgi.service.metatype.annotations.AttributeDefinition;
  68 import org.osgi.service.metatype.annotations.Designate;
  69 import org.osgi.service.metatype.annotations.ObjectClassDefinition;
  70 import org.slf4j.Logger;
  71 import org.slf4j.LoggerFactory;
  72
  73 @Singleton
  74 @Component(service = { }, configurationPid = "org.opendaylight.netconf.topology.singleton")
  75 @Designate(ocd = NetconfTopologyManager.Configuration.class)
  76 // Non-final for testing
  77 public class NetconfTopologyManager implements ClusteredDataTreeChangeListener<Node>, AutoCloseable {
  78     @ObjectClassDefinition
  79     public @interface Configuration {
  80         @AttributeDefinition(min = "1", description = "Name of the Network Topology instance to manage")
  81         String topology$_$id() default "topology-netconf";
  82
  83         @AttributeDefinition(min = "0", max = "65535",
  84             description = "Idle time in seconds after which write transaction is cancelled automatically. If 0, "
  85                 + "automatic cancellation is turned off.")
  86         int write$_$transaction$_$idle$_$timeout() default 0;
  87     }
  88
  89     private static final Logger LOG = LoggerFactory.getLogger(NetconfTopologyManager.class);
  90
  91     private final Map<InstanceIdentifier<Node>, NetconfTopologyContext> contexts = new ConcurrentHashMap<>();
  92     private final Map<InstanceIdentifier<Node>, ClusterSingletonServiceRegistration>
  93             clusterRegistrations = new ConcurrentHashMap<>();
  94
  95     private final BaseNetconfSchemas baseSchemas;
  96     private final DataBroker dataBroker;
  97     private final ClusterSingletonServiceProvider clusterSingletonServiceProvider;
  98     private final Timer timer;
  99     private final NetconfTopologySchemaAssembler schemaAssembler;
 100     private final ActorSystem actorSystem;
 101     private final NetconfClientFactory clientFactory;
 102     private final String topologyId;
 103     private final Duration writeTxIdleTimeout;
 104     private final DOMMountPointService mountPointService;
 105     private final DeviceActionFactory deviceActionFactory;
 106     private final NetconfClientConfigurationBuilderFactory builderFactory;
 107     private final SchemaResourceManager resourceManager;
 108
 109     private ListenerRegistration<NetconfTopologyManager> dataChangeListenerRegistration;
 110     private NetconfTopologyRPCProvider rpcProvider;
 111
 112     @Activate
 113     public NetconfTopologyManager(@Reference final BaseNetconfSchemas baseSchemas,
 114             @Reference final DataBroker dataBroker,
 115             @Reference final ClusterSingletonServiceProvider clusterSingletonServiceProvider,
 116             @Reference(target = "(type=global-timer)") final Timer timer,
 117             @Reference final NetconfTopologySchemaAssembler schemaAssembler,
 118             @Reference final ActorSystemProvider actorSystemProvider,
 119             @Reference(target = "(type=netconf-client-factory)") final NetconfClientFactory clientFactory,
 120             @Reference final DOMMountPointService mountPointService,
 121             @Reference final AAAEncryptionService encryptionService,
 122             @Reference final RpcProviderService rpcProviderService,
 123             @Reference final DeviceActionFactory deviceActionFactory,
 124             @Reference final SchemaResourceManager resourceManager,
 125             @Reference final NetconfClientConfigurationBuilderFactory builderFactory,
 126             final Configuration configuration) {
 127         this(baseSchemas, dataBroker, clusterSingletonServiceProvider, timer, schemaAssembler,
 128             actorSystemProvider.getActorSystem(), clientFactory, mountPointService, encryptionService,
 129             rpcProviderService, deviceActionFactory, resourceManager, builderFactory, configuration.topology$_$id(),
 130             Uint16.valueOf(configuration.write$_$transaction$_$idle$_$timeout()));
 131     }
 132
 133     @Inject
 134     public NetconfTopologyManager(final BaseNetconfSchemas baseSchemas, final DataBroker dataBroker,
 135             final ClusterSingletonServiceProvider clusterSingletonServiceProvider, final Timer timer,
 136             final NetconfTopologySchemaAssembler schemaAssembler, final ActorSystemProvider actorSystemProvider,
 137             final NetconfClientFactory clientFactory, final DOMMountPointService mountPointService,
 138             final AAAEncryptionService encryptionService, final RpcProviderService rpcProviderService,
 139             final DeviceActionFactory deviceActionFactory, final SchemaResourceManager resourceManager,
 140             final NetconfClientConfigurationBuilderFactory builderFactory) {
 141         this(baseSchemas, dataBroker, clusterSingletonServiceProvider, timer, schemaAssembler,
 142             actorSystemProvider.getActorSystem(), clientFactory, mountPointService, encryptionService,
 143             rpcProviderService, deviceActionFactory, resourceManager, builderFactory,
 144             NetconfNodeUtils.DEFAULT_TOPOLOGY_NAME, Uint16.ZERO);
 145     }
 146
 147     @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR",
 148         justification = "Non-final for mocking, but we register for DTCL and that leaks 'this'")
 149     public NetconfTopologyManager(final BaseNetconfSchemas baseSchemas, final DataBroker dataBroker,
 150             final ClusterSingletonServiceProvider clusterSingletonServiceProvider, final Timer timer,
 151             final NetconfTopologySchemaAssembler schemaAssembler, final ActorSystem actorSystem,
 152             final NetconfClientFactory clientFactory, final DOMMountPointService mountPointService,
 153             final AAAEncryptionService encryptionService, final RpcProviderService rpcProviderService,
 154             final DeviceActionFactory deviceActionFactory, final SchemaResourceManager resourceManager,
 155             final NetconfClientConfigurationBuilderFactory builderFactory, final String topologyId,
 156             final Uint16 writeTransactionIdleTimeout) {
 157         this.baseSchemas = requireNonNull(baseSchemas);
 158         this.dataBroker = requireNonNull(dataBroker);
 159         this.clusterSingletonServiceProvider = requireNonNull(clusterSingletonServiceProvider);
 160         this.timer = requireNonNull(timer);
 161         this.schemaAssembler = requireNonNull(schemaAssembler);
 162         this.actorSystem = requireNonNull(actorSystem);
 163         this.clientFactory = requireNonNull(clientFactory);
 164         this.topologyId = requireNonNull(topologyId);
 165         writeTxIdleTimeout = Duration.ofSeconds(writeTransactionIdleTimeout.toJava());
 166         this.mountPointService = mountPointService;
 167         this.deviceActionFactory = requireNonNull(deviceActionFactory);
 168         this.resourceManager = requireNonNull(resourceManager);
 169         this.builderFactory = requireNonNull(builderFactory);
 170
 171         dataChangeListenerRegistration = registerDataTreeChangeListener();
 172         rpcProvider = new NetconfTopologyRPCProvider(rpcProviderService, dataBroker, encryptionService, topologyId);
 173     }
 174
 175     @Override
 176     public void onDataTreeChanged(final Collection<DataTreeModification<Node>> changes) {
 177         for (final DataTreeModification<Node> change : changes) {
 178             final DataObjectModification<Node> rootNode = change.getRootNode();
 179             final InstanceIdentifier<Node> dataModifIdent = change.getRootPath().getRootIdentifier();
 180             final NodeId nodeId = NetconfTopologyUtils.getNodeId(rootNode.getIdentifier());
 181             switch (rootNode.getModificationType()) {
 182                 case SUBTREE_MODIFIED:
 183                     LOG.debug("Config for node {} updated", nodeId);
 184                     refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 185                     break;
 186                 case WRITE:
 187                     if (contexts.containsKey(dataModifIdent)) {
 188                         LOG.debug("RemoteDevice{{}} was already configured, reconfiguring node...", nodeId);
 189                         refreshNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 190                     } else {
 191                         LOG.debug("Config for node {} created", nodeId);
 192                         startNetconfDeviceContext(dataModifIdent, rootNode.getDataAfter());
 193                     }
 194                     break;
 195                 case DELETE:
 196                     LOG.debug("Config for node {} deleted", nodeId);
 197                     stopNetconfDeviceContext(dataModifIdent);
 198                     break;
 199                 default:
 200                     LOG.warn("Unknown operation for {}.", nodeId);
 201             }
 202         }
 203     }
 204
 205     private void refreshNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 206         final NetconfTopologyContext context = contexts.get(instanceIdentifier);
 207         context.refresh(createSetup(instanceIdentifier, node));
 208     }
 209
 210     // ClusterSingletonServiceRegistration registerClusterSingletonService method throws a Runtime exception if there
 211     // are problems with registration and client has to deal with it. Only thing we can do if this error occurs is to
 212     // retry registration several times and log the error.
 213     // TODO change to a specific documented Exception when changed in ClusterSingletonServiceProvider
 214     @SuppressWarnings("checkstyle:IllegalCatch")
 215     private void startNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 216         final NetconfNode netconfNode = requireNonNull(node.augmentation(NetconfNode.class));
 217
 218         final Timeout actorResponseWaitTime = Timeout.create(
 219                 Duration.ofSeconds(netconfNode.getActorResponseWaitTime().toJava()));
 220
 221         final ServiceGroupIdentifier serviceGroupIdent =
 222                 ServiceGroupIdentifier.create(instanceIdentifier.toString());
 223
 224         final NetconfTopologyContext newNetconfTopologyContext = newNetconfTopologyContext(
 225             createSetup(instanceIdentifier, node), serviceGroupIdent, actorResponseWaitTime, deviceActionFactory);
 226
 227         int tries = 3;
 228         while (true) {
 229             try {
 230                 final ClusterSingletonServiceRegistration clusterSingletonServiceRegistration =
 231                         clusterSingletonServiceProvider.registerClusterSingletonService(newNetconfTopologyContext);
 232                 clusterRegistrations.put(instanceIdentifier, clusterSingletonServiceRegistration);
 233                 contexts.put(instanceIdentifier, newNetconfTopologyContext);
 234                 break;
 235             } catch (final RuntimeException e) {
 236                 LOG.warn("Unable to register cluster singleton service {}, trying again", newNetconfTopologyContext, e);
 237
 238                 if (--tries <= 0) {
 239                     LOG.error("Unable to register cluster singleton service {} - done trying, closing topology context",
 240                             newNetconfTopologyContext, e);
 241                     close(newNetconfTopologyContext);
 242                     break;
 243                 }
 244             }
 245         }
 246     }
 247
 248     private void stopNetconfDeviceContext(final InstanceIdentifier<Node> instanceIdentifier) {
 249         final NetconfTopologyContext netconfTopologyContext = contexts.remove(instanceIdentifier);
 250         if (netconfTopologyContext != null) {
 251             close(clusterRegistrations.remove(instanceIdentifier));
 252             close(netconfTopologyContext);
 253         }
 254     }
 255
 256     @VisibleForTesting
 257     protected NetconfTopologyContext newNetconfTopologyContext(final NetconfTopologySetup setup,
 258             final ServiceGroupIdentifier serviceGroupIdent, final Timeout actorResponseWaitTime,
 259             final DeviceActionFactory deviceActionFact) {
 260         return new NetconfTopologyContext(resourceManager, mountPointService, builderFactory, deviceActionFactory,
 261             actorResponseWaitTime, serviceGroupIdent, setup);
 262     }
 263
 264     @PreDestroy
 265     @Deactivate
 266     @Override
 267     public void close() {
 268         if (rpcProvider != null) {
 269             rpcProvider.close();
 270             rpcProvider = null;
 271         }
 272         if (dataChangeListenerRegistration != null) {
 273             dataChangeListenerRegistration.close();
 274             dataChangeListenerRegistration = null;
 275         }
 276
 277         contexts.values().forEach(NetconfTopologyManager::close);
 278         clusterRegistrations.values().forEach(NetconfTopologyManager::close);
 279
 280         contexts.clear();
 281         clusterRegistrations.clear();
 282     }
 283
 284     @SuppressWarnings("checkstyle:IllegalCatch")
 285     private static void close(final AutoCloseable closeable) {
 286         try {
 287             closeable.close();
 288         } catch (Exception e) {
 289             LOG.warn("Error closing {}", closeable, e);
 290         }
 291     }
 292
 293     private ListenerRegistration<NetconfTopologyManager> registerDataTreeChangeListener() {
 294         final WriteTransaction wtx = dataBroker.newWriteOnlyTransaction();
 295         // FIXME: how does this play out with lifecycle? In a cluster, someone needs to ensure this call happens, but
 296         //        also we need to to make sure config -> oper is properly synchronized. Non-clustered case relies on
 297         //        oper being transient and perhaps on a put() instead, how do we handle that in the clustered case?
 298         wtx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.builder(NetworkTopology.class)
 299             .child(Topology.class, new TopologyKey(new TopologyId(topologyId)))
 300             .build(), new TopologyBuilder().setTopologyId(new TopologyId(topologyId)).build());
 301         wtx.commit().addCallback(new FutureCallback<CommitInfo>() {
 302             @Override
 303             public void onSuccess(final CommitInfo result) {
 304                 LOG.debug("topology initialization successful");
 305             }
 306
 307             @Override
 308             public void onFailure(final Throwable throwable) {
 309                 LOG.error("Unable to initialize netconf-topology", throwable);
 310             }
 311         }, MoreExecutors.directExecutor());
 312
 313         LOG.debug("Registering datastore listener");
 314         return dataBroker.registerDataTreeChangeListener(DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION,
 315             NetconfTopologyUtils.createTopologyListPath(topologyId).child(Node.class)), this);
 316     }
 317
 318     private NetconfTopologySetup createSetup(final InstanceIdentifier<Node> instanceIdentifier, final Node node) {
 319         final NetconfNode netconfNode = node.augmentation(NetconfNode.class);
 320         final RemoteDeviceId deviceId = NetconfNodeUtils.toRemoteDeviceId(node.getNodeId(), netconfNode);
 321
 322         return NetconfTopologySetup.builder()
 323             .setClusterSingletonServiceProvider(clusterSingletonServiceProvider)
 324             .setBaseSchemas(baseSchemas)
 325             .setDataBroker(dataBroker)
 326             .setInstanceIdentifier(instanceIdentifier)
 327             .setNode(node)
 328             .setActorSystem(actorSystem)
 329             .setTimer(timer)
 330             .setSchemaAssembler(schemaAssembler)
 331             .setTopologyId(topologyId)
 332             .setNetconfClientFactory(clientFactory)
 333             .setSchemaResourceDTO(resourceManager.getSchemaResources(netconfNode.getSchemaCacheDirectory(), deviceId))
 334             .setIdleTimeout(writeTxIdleTimeout)
 335             .build();
 336     }
 337 }