opendaylight/netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/impl/ClusteredNetconfTopology.java

   1 /*
   2  * Copyright (c) 2015 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.impl;
  10
  11 import akka.actor.ActorSystem;
  12 import akka.actor.TypedActor;
  13 import akka.actor.TypedActorExtension;
  14 import akka.actor.TypedProps;
  15 import akka.japi.Creator;
  16 import com.google.common.base.Optional;
  17 import com.google.common.base.Preconditions;
  18 import io.netty.util.concurrent.EventExecutor;
  19 import java.util.Collection;
  20 import java.util.Collections;
  21 import javassist.ClassPool;
  22 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  23 import org.opendaylight.controller.config.threadpool.ThreadPool;
  24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  25 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  26 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  27 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderContext;
  28 import org.opendaylight.controller.sal.core.api.Broker;
  29 import org.opendaylight.netconf.client.NetconfClientDispatcher;
  30 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
  31 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  32 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  33 import org.opendaylight.netconf.topology.AbstractNetconfTopology;
  34 import org.opendaylight.netconf.topology.NetconfTopology;
  35 import org.opendaylight.netconf.topology.NodeManagerCallback;
  36 import org.opendaylight.netconf.topology.NodeManagerCallback.NodeManagerCallbackFactory;
  37 import org.opendaylight.netconf.topology.SchemaRepositoryProvider;
  38 import org.opendaylight.netconf.topology.TopologyManager;
  39 import org.opendaylight.netconf.topology.TopologyManagerCallback;
  40 import org.opendaylight.netconf.topology.TopologyManagerCallback.TopologyManagerCallbackFactory;
  41 import org.opendaylight.netconf.topology.example.LoggingSalNodeWriter;
  42 import org.opendaylight.netconf.topology.pipeline.TopologyMountPointFacade;
  43 import org.opendaylight.netconf.topology.pipeline.TopologyMountPointFacade.ConnectionStatusListenerRegistration;
  44 import org.opendaylight.netconf.topology.util.BaseTopologyManager;
  45 import org.opendaylight.netconf.topology.util.NodeRoleChangeStrategy;
  46 import org.opendaylight.netconf.topology.util.NodeWriter;
  47 import org.opendaylight.netconf.topology.util.TopologyRoleChangeStrategy;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.$YangModuleInfoImpl;
  49 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  50 import org.opendaylight.yangtools.binding.data.codec.gen.impl.StreamWriterGenerator;
  51 import org.opendaylight.yangtools.binding.data.codec.impl.BindingNormalizedNodeCodecRegistry;
  52 import org.opendaylight.yangtools.sal.binding.generator.impl.ModuleInfoBackedContext;
  53 import org.opendaylight.yangtools.sal.binding.generator.util.BindingRuntimeContext;
  54 import org.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils;
  55 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  56 import org.slf4j.Logger;
  57 import org.slf4j.LoggerFactory;
  58
  59 public class ClusteredNetconfTopology extends AbstractNetconfTopology implements AutoCloseable {
  60
  61     private static final Logger LOG = LoggerFactory.getLogger(ClusteredNetconfTopology.class);
  62
  63     private final BindingNormalizedNodeCodecRegistry codecRegistry;
  64
  65     private final ActorSystem actorSystem;
  66     private final EntityOwnershipService entityOwnershipService;
  67     private TopologyManager topologyManager;
  68
  69     public ClusteredNetconfTopology(final String topologyId, final NetconfClientDispatcher clientDispatcher,
  70                                final BindingAwareBroker bindingAwareBroker, final Broker domBroker,
  71                                final EventExecutor eventExecutor, final ScheduledThreadPool keepaliveExecutor,
  72                                final ThreadPool processingExecutor, final SchemaRepositoryProvider schemaRepositoryProvider,
  73                                final ActorSystem actorSystem, final EntityOwnershipService entityOwnershipService) {
  74         super(topologyId, clientDispatcher,
  75                 bindingAwareBroker, domBroker, eventExecutor,
  76                 keepaliveExecutor, processingExecutor, schemaRepositoryProvider);
  77
  78         final ModuleInfoBackedContext moduleInfoBackedContext = ModuleInfoBackedContext.create();
  79         moduleInfoBackedContext.addModuleInfos(Collections.singletonList($YangModuleInfoImpl.getInstance()));
  80         final Optional<SchemaContext> schemaContextOptional = moduleInfoBackedContext.tryToCreateSchemaContext();
  81         Preconditions.checkState(schemaContextOptional.isPresent());
  82         final SchemaContext topologySchemaCtx = schemaContextOptional.get();
  83
  84         final JavassistUtils javassist = JavassistUtils.forClassPool(ClassPool.getDefault());
  85         codecRegistry = new BindingNormalizedNodeCodecRegistry(StreamWriterGenerator.create(javassist));
  86         codecRegistry.onBindingRuntimeContextUpdated(BindingRuntimeContext.create(moduleInfoBackedContext, topologySchemaCtx));
  87
  88         this.actorSystem = actorSystem;
  89         this.entityOwnershipService = entityOwnershipService;
  90         registerToSal(this, this);
  91         LOG.warn("Clustered netconf topo started");
  92     }
  93
  94     @Override
  95     public void onSessionInitiated(final ProviderContext session) {
  96         dataBroker = session.getSALService(DataBroker.class);
  97         final NodeWriter writer = new TopologyNodeWriter(topologyId, dataBroker);
  98         TypedActorExtension typedActorExtension = TypedActor.get(this.actorSystem);
  99         LOG.warn("Registering actor on path {}", actorSystem.name() + "/user/" + topologyId);
 100         topologyManager = typedActorExtension.typedActorOf(new TypedProps<>(TopologyManager.class, new Creator<BaseTopologyManager>() {
 101             @Override
 102             public BaseTopologyManager create() throws Exception {
 103                 return new BaseTopologyManager(actorSystem,
 104                         codecRegistry,
 105                         dataBroker,
 106                         topologyId,
 107                         new TopologyCallbackFactory(ClusteredNetconfTopology.this, entityOwnershipService, writer),
 108                         new NetconfNodeOperationalDataAggregator(),
 109                         new LoggingSalNodeWriter(writer),
 110                         new TopologyRoleChangeStrategy(dataBroker, entityOwnershipService, "topology-netconf", "topology-manager"));
 111             }
 112         }), topologyId);
 113     }
 114
 115     @Override
 116     public void close() throws Exception {
 117         // close all existing connectors, delete whole topology in datastore?
 118         for (NetconfConnectorDTO connectorDTO : activeConnectors.values()) {
 119             connectorDTO.getCommunicator().disconnect();
 120         }
 121         activeConnectors.clear();
 122     }
 123
 124     @Override
 125     protected RemoteDeviceHandler<NetconfSessionPreferences> createSalFacade(final RemoteDeviceId id, final Broker domBroker, final BindingAwareBroker bindingBroker, long defaultRequestTimeoutMillis) {
 126         return new TopologyMountPointFacade(id, domBroker, bindingBroker, defaultRequestTimeoutMillis);
 127     }
 128
 129     @Override
 130     public void registerMountPoint(NodeId nodeId) {
 131         ((TopologyMountPointFacade) activeConnectors.get(nodeId).getFacade()).registerMountPoint();
 132     }
 133
 134     @Override
 135     public void unregisterMountPoint(NodeId nodeId) {
 136         Preconditions.checkState(activeConnectors.containsKey(nodeId), "Cannot unregister nonexistent mountpoint");
 137         ((TopologyMountPointFacade) activeConnectors.get(nodeId).getFacade()).unregisterMountPoint();
 138     }
 139
 140     @Override
 141     public ConnectionStatusListenerRegistration registerConnectionStatusListener(final NodeId node, final RemoteDeviceHandler<NetconfSessionPreferences> listener) {
 142         Preconditions.checkState(activeConnectors.containsKey(node), "Need to connect a node before a connection listener can be registered");
 143         return ((TopologyMountPointFacade) activeConnectors.get(node).getFacade()).registerConnectionStatusListener(listener);
 144     }
 145
 146     @Override
 147     public Collection<ProviderFunctionality> getProviderFunctionality() {
 148         return Collections.emptySet();
 149     }
 150
 151     static class TopologyCallbackFactory implements TopologyManagerCallbackFactory {
 152
 153         private final NetconfTopology netconfTopology;
 154         private final EntityOwnershipService entityOwnershipService;
 155         private final NodeWriter writer;
 156
 157         TopologyCallbackFactory(final NetconfTopology netconfTopology, final EntityOwnershipService entityOwnershipService, final NodeWriter writer) {
 158             this.netconfTopology = netconfTopology;
 159             this.entityOwnershipService = entityOwnershipService;
 160             this.writer = writer;
 161         }
 162
 163         @Override
 164         public TopologyManagerCallback create(final ActorSystem actorSystem, final String topologyId) {
 165             return new NetconfTopologyManagerCallback(actorSystem, topologyId, new NodeCallbackFactory(netconfTopology, entityOwnershipService), new LoggingSalNodeWriter(writer));
 166         }
 167     }
 168
 169     private static class NodeCallbackFactory implements NodeManagerCallbackFactory {
 170
 171         private final NetconfTopology netconfTopology;
 172         private final EntityOwnershipService entityOwnershipService;
 173
 174         NodeCallbackFactory(final NetconfTopology netconfTopology, final EntityOwnershipService entityOwnershipService) {
 175             this.netconfTopology = netconfTopology;
 176             this.entityOwnershipService = entityOwnershipService;
 177         }
 178
 179         @Override
 180         public NodeManagerCallback create(final String nodeId, final String topologyId, final ActorSystem actorSystem) {
 181             return new NetconfNodeManagerCallback(nodeId, topologyId, actorSystem, netconfTopology, new NodeRoleChangeStrategy(entityOwnershipService, "netconf-node", nodeId));
 182         }
 183     }
 184 }