netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/AbstractNetconfTopology.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 package org.opendaylight.netconf.topology;
   9
  10 import com.google.common.base.Optional;
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.base.Strings;
  13 import com.google.common.collect.Lists;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import com.google.common.util.concurrent.ListeningExecutorService;
  18 import com.google.common.util.concurrent.MoreExecutors;
  19 import com.google.common.util.concurrent.Uninterruptibles;
  20 import io.netty.util.concurrent.EventExecutor;
  21 import java.io.File;
  22 import java.math.BigDecimal;
  23 import java.net.InetSocketAddress;
  24 import java.net.URL;
  25 import java.util.ArrayList;
  26 import java.util.HashMap;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.concurrent.TimeUnit;
  30 import org.opendaylight.aaa.encrypt.AAAEncryptionService;
  31 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  32 import org.opendaylight.controller.config.threadpool.ThreadPool;
  33 import org.opendaylight.mdsal.binding.api.DataBroker;
  34 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  35 import org.opendaylight.netconf.api.NetconfMessage;
  36 import org.opendaylight.netconf.client.NetconfClientDispatcher;
  37 import org.opendaylight.netconf.client.NetconfClientSessionListener;
  38 import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
  39 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfiguration;
  40 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder;
  41 import org.opendaylight.netconf.nettyutil.ReconnectStrategyFactory;
  42 import org.opendaylight.netconf.nettyutil.TimedReconnectStrategyFactory;
  43 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  44 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
  45 import org.opendaylight.netconf.sal.connect.api.DeviceActionFactory;
  46 import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
  47 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
  48 import org.opendaylight.netconf.sal.connect.netconf.LibraryModulesSchemas;
  49 import org.opendaylight.netconf.sal.connect.netconf.NetconfDevice;
  50 import org.opendaylight.netconf.sal.connect.netconf.NetconfDeviceBuilder;
  51 import org.opendaylight.netconf.sal.connect.netconf.NetconfStateSchemasResolverImpl;
  52 import org.opendaylight.netconf.sal.connect.netconf.SchemalessNetconfDevice;
  53 import org.opendaylight.netconf.sal.connect.netconf.auth.DatastoreBackedPublicKeyAuth;
  54 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCapabilities;
  55 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCommunicator;
  56 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  57 import org.opendaylight.netconf.sal.connect.netconf.listener.UserPreferences;
  58 import org.opendaylight.netconf.sal.connect.netconf.sal.KeepaliveSalFacade;
  59 import org.opendaylight.netconf.sal.connect.netconf.sal.NetconfKeystoreAdapter;
  60 import org.opendaylight.netconf.sal.connect.netconf.schema.YangLibrarySchemaYangSourceProvider;
  61 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  62 import org.opendaylight.netconf.sal.connect.util.SslHandlerFactoryImpl;
  63 import org.opendaylight.netconf.topology.api.NetconfTopology;
  64 import org.opendaylight.netconf.topology.api.SchemaRepositoryProvider;
  65 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
  66 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.Protocol;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.Protocol.Name;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapability.CapabilityOrigin;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.Credentials;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.KeyAuth;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.LoginPw;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.LoginPwUnencrypted;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.key.auth.KeyBased;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.login.pw.LoginPassword;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.login.pw.unencrypted.LoginPasswordUnencrypted;
  78 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  79 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  80 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
  81 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
  82 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter;
  83 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  84 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  85 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  86 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
  87 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
  88 import org.opendaylight.yangtools.yang.model.repo.util.FilesystemSchemaSourceCache;
  89 import org.opendaylight.yangtools.yang.model.repo.util.InMemorySchemaSourceCache;
  90 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
  91 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.ASTSchemaSource;
  92 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToASTTransformer;
  93 import org.slf4j.Logger;
  94 import org.slf4j.LoggerFactory;
  95
  96 public abstract class AbstractNetconfTopology implements NetconfTopology {
  97
  98     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfTopology.class);
  99
 100     protected static final long DEFAULT_REQUEST_TIMEOUT_MILLIS = 60000L;
 101     protected static final int DEFAULT_KEEPALIVE_DELAY = 0;
 102     protected static final boolean DEFAULT_RECONNECT_ON_CHANGED_SCHEMA = false;
 103     protected static final int DEFAULT_CONCURRENT_RPC_LIMIT = 0;
 104     private static final int DEFAULT_MAX_CONNECTION_ATTEMPTS = 0;
 105     private static final int DEFAULT_BETWEEN_ATTEMPTS_TIMEOUT_MILLIS = 2000;
 106     private static final long DEFAULT_CONNECTION_TIMEOUT_MILLIS = 20000L;
 107     private static final BigDecimal DEFAULT_SLEEP_FACTOR = new BigDecimal(1.5);
 108
 109     // constants related to Schema Cache(s)
 110     /**
 111      * Filesystem based caches are stored relative to the cache directory.
 112      */
 113     private static final String CACHE_DIRECTORY = "cache";
 114
 115     /**
 116      * The default cache directory relative to <code>CACHE_DIRECTORY</code>.
 117      */
 118     private static final String DEFAULT_CACHE_DIRECTORY = "schema";
 119
 120     /**
 121      * The qualified schema cache directory <code>cache/schema</code>.
 122      */
 123     private static final String QUALIFIED_DEFAULT_CACHE_DIRECTORY =
 124             CACHE_DIRECTORY + File.separator + DEFAULT_CACHE_DIRECTORY;
 125
 126     /**
 127      * The name for the default schema repository.
 128      */
 129     private static final String DEFAULT_SCHEMA_REPOSITORY_NAME = "sal-netconf-connector";
 130
 131     /**
 132      * The default schema repository in the case that one is not specified.
 133      */
 134     private static final SharedSchemaRepository DEFAULT_SCHEMA_REPOSITORY =
 135             new SharedSchemaRepository(DEFAULT_SCHEMA_REPOSITORY_NAME);
 136
 137     public static final InMemorySchemaSourceCache<ASTSchemaSource> DEFAULT_AST_CACHE =
 138             InMemorySchemaSourceCache.createSoftCache(DEFAULT_SCHEMA_REPOSITORY, ASTSchemaSource.class);
 139
 140     /**
 141      * The default factory for creating <code>SchemaContext</code> instances.
 142      */
 143     private static final SchemaContextFactory DEFAULT_SCHEMA_CONTEXT_FACTORY =
 144             DEFAULT_SCHEMA_REPOSITORY.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT);
 145
 146     /**
 147      * Keeps track of initialized Schema resources.  A Map is maintained in which the key represents the name
 148      * of the schema cache directory, and the value is a corresponding <code>SchemaResourcesDTO</code>.  The
 149      * <code>SchemaResourcesDTO</code> is essentially a container that allows for the extraction of the
 150      * <code>SchemaRegistry</code> and <code>SchemaContextFactory</code> which should be used for a particular
 151      * Netconf mount.  Access to <code>SCHEMA_RESOURCES_DTO_MAP</code> should be surrounded by appropriate
 152      * synchronization locks.
 153      */
 154     private static final Map<String, NetconfDevice.SchemaResourcesDTO> SCHEMA_RESOURCES_DTO_MAP = new HashMap<>();
 155
 156     // Initializes default constant instances for the case when the default schema repository
 157     // directory cache/schema is used.
 158     static {
 159         SCHEMA_RESOURCES_DTO_MAP.put(DEFAULT_CACHE_DIRECTORY,
 160                 new NetconfDevice.SchemaResourcesDTO(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY,
 161                         DEFAULT_SCHEMA_CONTEXT_FACTORY,
 162                         new NetconfStateSchemasResolverImpl()));
 163         DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(DEFAULT_AST_CACHE);
 164         DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(
 165                 TextToASTTransformer.create(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY));
 166
 167         /*
 168          * Create the default <code>FilesystemSchemaSourceCache</code>, which stores cached files
 169          * in <code>cache/schema</code>. Try up to 3 times - we've seen intermittent failures on jenkins where
 170          * FilesystemSchemaSourceCache throws an IAE due to mkdirs failure. The theory is that there's a race
 171          * creating the dir and it already exists when mkdirs is called (mkdirs returns false in this case). In this
 172          * scenario, a retry should succeed.
 173          */
 174         int tries = 1;
 175         while (true) {
 176             try {
 177                 FilesystemSchemaSourceCache<YangTextSchemaSource> defaultCache =
 178                         new FilesystemSchemaSourceCache<>(DEFAULT_SCHEMA_REPOSITORY, YangTextSchemaSource.class,
 179                                 new File(QUALIFIED_DEFAULT_CACHE_DIRECTORY));
 180                 DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(defaultCache);
 181                 break;
 182             } catch (IllegalArgumentException e) {
 183                 if (tries++ >= 3) {
 184                     LOG.error("Error creating default schema cache", e);
 185                     break;
 186                 }
 187                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
 188             }
 189         }
 190     }
 191
 192     private final NetconfClientDispatcher clientDispatcher;
 193     private final EventExecutor eventExecutor;
 194     private final DeviceActionFactory deviceActionFactory;
 195     private final NetconfKeystoreAdapter keystoreAdapter;
 196     protected final ScheduledThreadPool keepaliveExecutor;
 197     protected final ListeningExecutorService processingExecutor;
 198     protected final SharedSchemaRepository sharedSchemaRepository;
 199     protected final DataBroker dataBroker;
 200     protected final DOMMountPointService mountPointService;
 201     protected final String topologyId;
 202     protected SchemaSourceRegistry schemaRegistry = DEFAULT_SCHEMA_REPOSITORY;
 203     protected SchemaRepository schemaRepository = DEFAULT_SCHEMA_REPOSITORY;
 204     protected SchemaContextFactory schemaContextFactory = DEFAULT_SCHEMA_CONTEXT_FACTORY;
 205     protected String privateKeyPath;
 206     protected String privateKeyPassphrase;
 207     protected final AAAEncryptionService encryptionService;
 208     protected final HashMap<NodeId, NetconfConnectorDTO> activeConnectors = new HashMap<>();
 209
 210     protected AbstractNetconfTopology(final String topologyId, final NetconfClientDispatcher clientDispatcher,
 211                                       final EventExecutor eventExecutor, final ScheduledThreadPool keepaliveExecutor,
 212                                       final ThreadPool processingExecutor,
 213                                       final SchemaRepositoryProvider schemaRepositoryProvider,
 214                                       final DataBroker dataBroker, final DOMMountPointService mountPointService,
 215                                       final AAAEncryptionService encryptionService,
 216                                       final DeviceActionFactory deviceActionFactory) {
 217         this.topologyId = topologyId;
 218         this.clientDispatcher = clientDispatcher;
 219         this.eventExecutor = eventExecutor;
 220         this.keepaliveExecutor = keepaliveExecutor;
 221         this.processingExecutor = MoreExecutors.listeningDecorator(processingExecutor.getExecutor());
 222         this.deviceActionFactory = deviceActionFactory;
 223         this.sharedSchemaRepository = schemaRepositoryProvider.getSharedSchemaRepository();
 224         this.dataBroker = dataBroker;
 225         this.mountPointService = mountPointService;
 226         this.encryptionService = encryptionService;
 227
 228         this.keystoreAdapter = new NetconfKeystoreAdapter(dataBroker);
 229     }
 230
 231     public void setSchemaRegistry(final SchemaSourceRegistry schemaRegistry) {
 232         this.schemaRegistry = schemaRegistry;
 233     }
 234
 235     public void setSchemaContextFactory(final SchemaContextFactory schemaContextFactory) {
 236         this.schemaContextFactory = schemaContextFactory;
 237     }
 238
 239     @Override
 240     public ListenableFuture<NetconfDeviceCapabilities> connectNode(final NodeId nodeId, final Node configNode) {
 241         LOG.info("Connecting RemoteDevice{{}} , with config {}", nodeId, configNode);
 242         return setupConnection(nodeId, configNode);
 243     }
 244
 245     @Override
 246     public ListenableFuture<Void> disconnectNode(final NodeId nodeId) {
 247         LOG.debug("Disconnecting RemoteDevice{{}}", nodeId.getValue());
 248         if (!activeConnectors.containsKey(nodeId)) {
 249             return Futures.immediateFailedFuture(
 250                     new IllegalStateException("Unable to disconnect device that is not connected"));
 251         }
 252
 253         // retrieve connection, and disconnect it
 254         final NetconfConnectorDTO connectorDTO = activeConnectors.remove(nodeId);
 255         connectorDTO.getCommunicator().close();
 256         connectorDTO.getFacade().close();
 257         return Futures.immediateFuture(null);
 258     }
 259
 260     protected ListenableFuture<NetconfDeviceCapabilities> setupConnection(final NodeId nodeId,
 261                                                                           final Node configNode) {
 262         final NetconfNode netconfNode = configNode.augmentation(NetconfNode.class);
 263
 264         Preconditions.checkNotNull(netconfNode.getHost());
 265         Preconditions.checkNotNull(netconfNode.getPort());
 266         Preconditions.checkNotNull(netconfNode.isTcpOnly());
 267
 268         final NetconfConnectorDTO deviceCommunicatorDTO = createDeviceCommunicator(nodeId, netconfNode);
 269         final NetconfDeviceCommunicator deviceCommunicator = deviceCommunicatorDTO.getCommunicator();
 270         final NetconfClientSessionListener netconfClientSessionListener = deviceCommunicatorDTO.getSessionListener();
 271         final NetconfReconnectingClientConfiguration clientConfig =
 272                 getClientConfig(netconfClientSessionListener, netconfNode);
 273         final ListenableFuture<NetconfDeviceCapabilities> future =
 274                 deviceCommunicator.initializeRemoteConnection(clientDispatcher, clientConfig);
 275
 276         activeConnectors.put(nodeId, deviceCommunicatorDTO);
 277
 278         Futures.addCallback(future, new FutureCallback<NetconfDeviceCapabilities>() {
 279             @Override
 280             public void onSuccess(final NetconfDeviceCapabilities result) {
 281                 LOG.debug("Connector for {} started succesfully", nodeId.getValue());
 282             }
 283
 284             @Override
 285             public void onFailure(final Throwable throwable) {
 286                 LOG.error("Connector for {} failed", nodeId.getValue(), throwable);
 287                 // remove this node from active connectors?
 288             }
 289         }, MoreExecutors.directExecutor());
 290
 291         return future;
 292     }
 293
 294     protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId, final NetconfNode node) {
 295         //setup default values since default value is not supported in mdsal
 296         final long defaultRequestTimeoutMillis = node.getDefaultRequestTimeoutMillis() == null
 297                 ? DEFAULT_REQUEST_TIMEOUT_MILLIS : node.getDefaultRequestTimeoutMillis();
 298         final long keepaliveDelay = node.getKeepaliveDelay() == null
 299                 ? DEFAULT_KEEPALIVE_DELAY : node.getKeepaliveDelay();
 300         final boolean reconnectOnChangedSchema = node.isReconnectOnChangedSchema() == null
 301                 ? DEFAULT_RECONNECT_ON_CHANGED_SCHEMA : node.isReconnectOnChangedSchema();
 302
 303         final IpAddress ipAddress = node.getHost().getIpAddress();
 304         final InetSocketAddress address = new InetSocketAddress(ipAddress.getIpv4Address() != null
 305                 ? ipAddress.getIpv4Address().getValue() : ipAddress.getIpv6Address().getValue(),
 306                 node.getPort().getValue());
 307         final RemoteDeviceId remoteDeviceId = new RemoteDeviceId(nodeId.getValue(), address);
 308
 309         RemoteDeviceHandler<NetconfSessionPreferences> salFacade =
 310                 createSalFacade(remoteDeviceId);
 311
 312         if (keepaliveDelay > 0) {
 313             LOG.warn("Adding keepalive facade, for device {}", nodeId);
 314             salFacade = new KeepaliveSalFacade(remoteDeviceId, salFacade, this.keepaliveExecutor.getExecutor(),
 315                     keepaliveDelay, defaultRequestTimeoutMillis);
 316         }
 317
 318         // pre register yang library sources as fallback schemas to schema registry
 319         final List<SchemaSourceRegistration<YangTextSchemaSource>> registeredYangLibSources = Lists.newArrayList();
 320         if (node.getYangLibrary() != null) {
 321             final String yangLibURL = node.getYangLibrary().getYangLibraryUrl().getValue();
 322             final String yangLibUsername = node.getYangLibrary().getUsername();
 323             final String yangLigPassword = node.getYangLibrary().getPassword();
 324
 325             final LibraryModulesSchemas libraryModulesSchemas;
 326             if (yangLibURL != null) {
 327                 if (yangLibUsername != null && yangLigPassword != null) {
 328                     libraryModulesSchemas = LibraryModulesSchemas.create(yangLibURL, yangLibUsername, yangLigPassword);
 329                 } else {
 330                     libraryModulesSchemas = LibraryModulesSchemas.create(yangLibURL);
 331                 }
 332
 333                 for (final Map.Entry<SourceIdentifier, URL> sourceIdentifierURLEntry
 334                         : libraryModulesSchemas.getAvailableModels().entrySet()) {
 335                     registeredYangLibSources
 336                         .add(schemaRegistry.registerSchemaSource(
 337                                 new YangLibrarySchemaYangSourceProvider(remoteDeviceId,
 338                                         libraryModulesSchemas.getAvailableModels()),
 339                                 PotentialSchemaSource.create(sourceIdentifierURLEntry.getKey(),
 340                                         YangTextSchemaSource.class, PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
 341                 }
 342             }
 343         }
 344
 345         final NetconfDevice.SchemaResourcesDTO schemaResourcesDTO = setupSchemaCacheDTO(nodeId, node);
 346         final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> device;
 347         if (node.isSchemaless()) {
 348             device = new SchemalessNetconfDevice(remoteDeviceId, salFacade);
 349         } else {
 350             NetconfDeviceBuilder netconfDeviceBuilder = new NetconfDeviceBuilder()
 351                     .setReconnectOnSchemasChange(reconnectOnChangedSchema)
 352                     .setSchemaResourcesDTO(schemaResourcesDTO)
 353                     .setGlobalProcessingExecutor(this.processingExecutor)
 354                     .setId(remoteDeviceId)
 355                     .setSalFacade(salFacade);
 356             if (this.deviceActionFactory != null) {
 357                 netconfDeviceBuilder.setDeviceActionFactory(this.deviceActionFactory);
 358             }
 359             device = netconfDeviceBuilder.build();
 360         }
 361
 362         final Optional<UserPreferences> userCapabilities = getUserCapabilities(node);
 363         final int rpcMessageLimit =
 364                 node.getConcurrentRpcLimit() == null ? DEFAULT_CONCURRENT_RPC_LIMIT : node.getConcurrentRpcLimit();
 365
 366         if (rpcMessageLimit < 1) {
 367             LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", remoteDeviceId);
 368         }
 369
 370         NetconfDeviceCommunicator netconfDeviceCommunicator =
 371              userCapabilities.isPresent() ? new NetconfDeviceCommunicator(remoteDeviceId, device,
 372                      userCapabilities.get(), rpcMessageLimit)
 373             : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit);
 374
 375         if (salFacade instanceof KeepaliveSalFacade) {
 376             ((KeepaliveSalFacade)salFacade).setListener(netconfDeviceCommunicator);
 377         }
 378         return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade);
 379     }
 380
 381     protected NetconfDevice.SchemaResourcesDTO setupSchemaCacheDTO(final NodeId nodeId, final NetconfNode node) {
 382         // Setup information related to the SchemaRegistry, SchemaResourceFactory, etc.
 383         NetconfDevice.SchemaResourcesDTO schemaResourcesDTO = null;
 384         final String moduleSchemaCacheDirectory = node.getSchemaCacheDirectory();
 385         // Only checks to ensure the String is not empty or null; further checks related to directory
 386         // accessibility and file permissionsare handled during the FilesystemSchemaSourceCache initialization.
 387         if (!Strings.isNullOrEmpty(moduleSchemaCacheDirectory)) {
 388             // If a custom schema cache directory is specified, create the backing DTO; otherwise,
 389             // the SchemaRegistry and SchemaContextFactory remain the default values.
 390             if (!moduleSchemaCacheDirectory.equals(DEFAULT_CACHE_DIRECTORY)) {
 391                 // Multiple modules may be created at once;
 392                 // synchronize to avoid issues with data consistency among threads.
 393                 synchronized (SCHEMA_RESOURCES_DTO_MAP) {
 394                     // Look for the cached DTO to reuse SchemaRegistry and SchemaContextFactory variables
 395                     // if they already exist
 396                     schemaResourcesDTO = SCHEMA_RESOURCES_DTO_MAP.get(moduleSchemaCacheDirectory);
 397                     if (schemaResourcesDTO == null) {
 398                         schemaResourcesDTO = createSchemaResourcesDTO(moduleSchemaCacheDirectory);
 399                         schemaResourcesDTO.getSchemaRegistry().registerSchemaSourceListener(
 400                                 TextToASTTransformer.create((SchemaRepository) schemaResourcesDTO.getSchemaRegistry(),
 401                                         schemaResourcesDTO.getSchemaRegistry())
 402                         );
 403                         SCHEMA_RESOURCES_DTO_MAP.put(moduleSchemaCacheDirectory, schemaResourcesDTO);
 404                     }
 405                 }
 406                 LOG.info("Netconf connector for device {} will use schema cache directory {} instead of {}",
 407                         nodeId.getValue(), moduleSchemaCacheDirectory, DEFAULT_CACHE_DIRECTORY);
 408             }
 409         } else {
 410             LOG.warn("schema-cache-directory for {} is null or empty;  using the default {}",
 411                     nodeId.getValue(), QUALIFIED_DEFAULT_CACHE_DIRECTORY);
 412         }
 413
 414         if (schemaResourcesDTO == null) {
 415             schemaResourcesDTO = new NetconfDevice.SchemaResourcesDTO(schemaRegistry, schemaRepository,
 416                     schemaContextFactory, new NetconfStateSchemasResolverImpl());
 417         }
 418
 419         return schemaResourcesDTO;
 420     }
 421
 422     /**
 423      * Creates the backing Schema classes for a particular directory.
 424      *
 425      * @param moduleSchemaCacheDirectory The string directory relative to "cache"
 426      * @return A DTO containing the Schema classes for the Netconf mount.
 427      */
 428     private NetconfDevice.SchemaResourcesDTO createSchemaResourcesDTO(final String moduleSchemaCacheDirectory) {
 429         final SharedSchemaRepository repository = new SharedSchemaRepository(moduleSchemaCacheDirectory);
 430         final SchemaContextFactory contextFactory
 431                 = repository.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT);
 432         setSchemaRegistry(repository);
 433         setSchemaContextFactory(contextFactory);
 434         final FilesystemSchemaSourceCache<YangTextSchemaSource> deviceCache =
 435                 createDeviceFilesystemCache(moduleSchemaCacheDirectory);
 436         repository.registerSchemaSourceListener(deviceCache);
 437         repository.registerSchemaSourceListener(
 438             InMemorySchemaSourceCache.createSoftCache(repository, ASTSchemaSource.class));
 439         return new NetconfDevice.SchemaResourcesDTO(repository, repository, contextFactory,
 440                 new NetconfStateSchemasResolverImpl());
 441     }
 442
 443     /**
 444      * Creates a <code>FilesystemSchemaSourceCache</code> for the custom schema cache directory.
 445      *
 446      * @param schemaCacheDirectory The custom cache directory relative to "cache"
 447      * @return A <code>FilesystemSchemaSourceCache</code> for the custom schema cache directory
 448      */
 449     private FilesystemSchemaSourceCache<YangTextSchemaSource> createDeviceFilesystemCache(
 450             final String schemaCacheDirectory) {
 451         final String relativeSchemaCacheDirectory = CACHE_DIRECTORY + File.separator + schemaCacheDirectory;
 452         return new FilesystemSchemaSourceCache<>(schemaRegistry, YangTextSchemaSource.class,
 453                 new File(relativeSchemaCacheDirectory));
 454     }
 455
 456     /**
 457      * Sets the private key path from location specified in configuration file using blueprint.
 458      */
 459     public void setPrivateKeyPath(final String privateKeyPath) {
 460         this.privateKeyPath = privateKeyPath;
 461     }
 462
 463     /**
 464      * Sets the private key passphrase from location specified in configuration file using blueprint.
 465      */
 466     public void setPrivateKeyPassphrase(final String privateKeyPassphrase) {
 467         this.privateKeyPassphrase = privateKeyPassphrase;
 468     }
 469
 470     public NetconfReconnectingClientConfiguration getClientConfig(final NetconfClientSessionListener listener,
 471                                                                   final NetconfNode node) {
 472
 473         //setup default values since default value is not supported in mdsal
 474         final long clientConnectionTimeoutMillis = node.getConnectionTimeoutMillis() == null
 475                 ? DEFAULT_CONNECTION_TIMEOUT_MILLIS : node.getConnectionTimeoutMillis();
 476         final long maxConnectionAttempts = node.getMaxConnectionAttempts() == null
 477                 ? DEFAULT_MAX_CONNECTION_ATTEMPTS : node.getMaxConnectionAttempts();
 478         final int betweenAttemptsTimeoutMillis = node.getBetweenAttemptsTimeoutMillis() == null
 479                 ? DEFAULT_BETWEEN_ATTEMPTS_TIMEOUT_MILLIS : node.getBetweenAttemptsTimeoutMillis();
 480         final BigDecimal sleepFactor = node.getSleepFactor() == null ? DEFAULT_SLEEP_FACTOR : node.getSleepFactor();
 481
 482         final InetSocketAddress socketAddress = getSocketAddress(node.getHost(), node.getPort().getValue());
 483
 484         final ReconnectStrategyFactory sf = new TimedReconnectStrategyFactory(eventExecutor,
 485                 maxConnectionAttempts, betweenAttemptsTimeoutMillis, sleepFactor);
 486
 487         final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder;
 488         final Protocol protocol = node.getProtocol();
 489         if (node.isTcpOnly()) {
 490             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 491                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TCP)
 492                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 493         } else if (protocol == null || protocol.getName() == Name.SSH) {
 494             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 495                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH)
 496                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 497         } else if (protocol.getName() == Name.TLS) {
 498             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 499                 .withSslHandlerFactory(new SslHandlerFactoryImpl(keystoreAdapter, protocol.getSpecification()))
 500                 .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
 501         } else {
 502             throw new IllegalStateException("Unsupported protocol type: " + protocol.getName());
 503         }
 504
 505         if (node.getOdlHelloMessageCapabilities() != null) {
 506             reconnectingClientConfigurationBuilder
 507                     .withOdlHelloCapabilities(node.getOdlHelloMessageCapabilities().getCapability());
 508         }
 509
 510         return reconnectingClientConfigurationBuilder
 511                 .withAddress(socketAddress)
 512                 .withConnectionTimeoutMillis(clientConnectionTimeoutMillis)
 513                 .withReconnectStrategy(sf.createReconnectStrategy())
 514                 .withConnectStrategyFactory(sf)
 515                 .withSessionListener(listener)
 516                 .build();
 517     }
 518
 519     private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
 520         if (credentials instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 521                 .rev150114.netconf.node.credentials.credentials.LoginPassword) {
 522             final org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 523                     .rev150114.netconf.node.credentials.credentials.LoginPassword loginPassword
 524                     = (org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 525                     .rev150114.netconf.node.credentials.credentials.LoginPassword) credentials;
 526             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 527         }
 528         if (credentials instanceof LoginPwUnencrypted) {
 529             final LoginPasswordUnencrypted loginPassword =
 530                     ((LoginPwUnencrypted) credentials).getLoginPasswordUnencrypted();
 531             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 532         }
 533         if (credentials instanceof LoginPw) {
 534             final LoginPassword loginPassword = ((LoginPw) credentials).getLoginPassword();
 535             return new LoginPasswordHandler(loginPassword.getUsername(),
 536                     encryptionService.decrypt(loginPassword.getPassword()));
 537         }
 538         if (credentials instanceof KeyAuth) {
 539             final KeyBased keyPair = ((KeyAuth) credentials).getKeyBased();
 540             return new DatastoreBackedPublicKeyAuth(keyPair.getUsername(), keyPair.getKeyId(),
 541                     keystoreAdapter, encryptionService);
 542         }
 543         throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
 544     }
 545
 546     protected abstract RemoteDeviceHandler<NetconfSessionPreferences> createSalFacade(RemoteDeviceId id);
 547
 548     private static InetSocketAddress getSocketAddress(final Host host, final int port) {
 549         if (host.getDomainName() != null) {
 550             return new InetSocketAddress(host.getDomainName().getValue(), port);
 551         }
 552
 553         final IpAddress ipAddress = host.getIpAddress();
 554         final String ip = ipAddress.getIpv4Address() != null ? ipAddress.getIpv4Address().getValue()
 555                 : ipAddress.getIpv6Address().getValue();
 556         return new InetSocketAddress(ip, port);
 557     }
 558
 559     private static Optional<UserPreferences> getUserCapabilities(final NetconfNode node) {
 560         // if none of yang-module-capabilities or non-module-capabilities is specified
 561         // just return absent
 562         if (node.getYangModuleCapabilities() == null && node.getNonModuleCapabilities() == null) {
 563             return Optional.absent();
 564         }
 565
 566         final List<String> capabilities = new ArrayList<>();
 567
 568         boolean overrideYangModuleCaps = false;
 569         if (node.getYangModuleCapabilities() != null) {
 570             capabilities.addAll(node.getYangModuleCapabilities().getCapability());
 571             overrideYangModuleCaps = node.getYangModuleCapabilities().isOverride();
 572         }
 573
 574         //non-module capabilities should not exist in yang module capabilities
 575         final NetconfSessionPreferences netconfSessionPreferences = NetconfSessionPreferences.fromStrings(capabilities);
 576         Preconditions.checkState(netconfSessionPreferences.getNonModuleCaps().isEmpty(),
 577                 "List yang-module-capabilities/capability should contain only module based capabilities. "
 578                         + "Non-module capabilities used: " + netconfSessionPreferences.getNonModuleCaps());
 579
 580         boolean overrideNonModuleCaps = false;
 581         if (node.getNonModuleCapabilities() != null) {
 582             capabilities.addAll(node.getNonModuleCapabilities().getCapability());
 583             overrideNonModuleCaps = node.getNonModuleCapabilities().isOverride();
 584         }
 585
 586         return Optional.of(new UserPreferences(NetconfSessionPreferences
 587             .fromStrings(capabilities, CapabilityOrigin.UserDefined), overrideYangModuleCaps, overrideNonModuleCaps));
 588     }
 589
 590     protected static class NetconfConnectorDTO implements AutoCloseable {
 591
 592         private final NetconfDeviceCommunicator communicator;
 593         private final RemoteDeviceHandler<NetconfSessionPreferences> facade;
 594
 595         public NetconfConnectorDTO(final NetconfDeviceCommunicator communicator,
 596                                    final RemoteDeviceHandler<NetconfSessionPreferences> facade) {
 597             this.communicator = communicator;
 598             this.facade = facade;
 599         }
 600
 601         public NetconfDeviceCommunicator getCommunicator() {
 602             return communicator;
 603         }
 604
 605         public RemoteDeviceHandler<NetconfSessionPreferences> getFacade() {
 606             return facade;
 607         }
 608
 609         public NetconfClientSessionListener getSessionListener() {
 610             return communicator;
 611         }
 612
 613         @Override
 614         public void close() {
 615             communicator.close();
 616             facade.close();
 617         }
 618     }
 619 }