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