import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.EventExecutor;
import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder;
import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
+import org.opendaylight.netconf.sal.connect.api.DeviceActionFactory;
import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
import org.opendaylight.netconf.sal.connect.netconf.LibraryModulesSchemas;
import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
import org.opendaylight.yangtools.yang.model.repo.util.FilesystemSchemaSourceCache;
+import org.opendaylight.yangtools.yang.model.repo.util.InMemorySchemaSourceCache;
import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
+import org.opendaylight.yangtools.yang.parser.rfc7950.repo.ASTSchemaSource;
import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToASTTransformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private static final SharedSchemaRepository DEFAULT_SCHEMA_REPOSITORY =
new SharedSchemaRepository(DEFAULT_SCHEMA_REPOSITORY_NAME);
- /**
- * The default <code>FilesystemSchemaSourceCache</code>, which stores cached files in <code>cache/schema</code>.
- */
- private static final FilesystemSchemaSourceCache<YangTextSchemaSource> DEFAULT_CACHE =
- new FilesystemSchemaSourceCache<>(DEFAULT_SCHEMA_REPOSITORY, YangTextSchemaSource.class,
- new File(QUALIFIED_DEFAULT_CACHE_DIRECTORY));
+ public static final InMemorySchemaSourceCache<ASTSchemaSource> DEFAULT_AST_CACHE =
+ InMemorySchemaSourceCache.createSoftCache(DEFAULT_SCHEMA_REPOSITORY, ASTSchemaSource.class);
/**
* The default factory for creating <code>SchemaContext</code> instances.
new NetconfDevice.SchemaResourcesDTO(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY,
DEFAULT_SCHEMA_CONTEXT_FACTORY,
new NetconfStateSchemasResolverImpl()));
- DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(DEFAULT_CACHE);
+ DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(DEFAULT_AST_CACHE);
DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(
TextToASTTransformer.create(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY));
+
+ /*
+ * Create the default <code>FilesystemSchemaSourceCache</code>, which stores cached files
+ * in <code>cache/schema</code>. Try up to 3 times - we've seen intermittent failures on jenkins where
+ * FilesystemSchemaSourceCache throws an IAE due to mkdirs failure. The theory is that there's a race
+ * creating the dir and it already exists when mkdirs is called (mkdirs returns false in this case). In this
+ * scenario, a retry should succeed.
+ */
+ int tries = 1;
+ while (true) {
+ try {
+ FilesystemSchemaSourceCache<YangTextSchemaSource> defaultCache =
+ new FilesystemSchemaSourceCache<>(DEFAULT_SCHEMA_REPOSITORY, YangTextSchemaSource.class,
+ new File(QUALIFIED_DEFAULT_CACHE_DIRECTORY));
+ DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(defaultCache);
+ break;
+ } catch (IllegalArgumentException e) {
+ if (tries++ >= 3) {
+ LOG.error("Error creating default schema cache", e);
+ break;
+ }
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ }
}
- protected final String topologyId;
private final NetconfClientDispatcher clientDispatcher;
private final EventExecutor eventExecutor;
+ private final DeviceActionFactory deviceActionFactory;
+ private final NetconfKeystoreAdapter keystoreAdapter;
protected final ScheduledThreadPool keepaliveExecutor;
protected final ThreadPool processingExecutor;
protected final SharedSchemaRepository sharedSchemaRepository;
protected final DataBroker dataBroker;
protected final DOMMountPointService mountPointService;
- private final NetconfKeystoreAdapter keystoreAdapter;
+ protected final String topologyId;
protected SchemaSourceRegistry schemaRegistry = DEFAULT_SCHEMA_REPOSITORY;
protected SchemaRepository schemaRepository = DEFAULT_SCHEMA_REPOSITORY;
protected SchemaContextFactory schemaContextFactory = DEFAULT_SCHEMA_CONTEXT_FACTORY;
final ThreadPool processingExecutor,
final SchemaRepositoryProvider schemaRepositoryProvider,
final DataBroker dataBroker, final DOMMountPointService mountPointService,
- final AAAEncryptionService encryptionService) {
+ final AAAEncryptionService encryptionService,
+ final DeviceActionFactory deviceActionFactory) {
this.topologyId = topologyId;
this.clientDispatcher = clientDispatcher;
this.eventExecutor = eventExecutor;
this.keepaliveExecutor = keepaliveExecutor;
this.processingExecutor = processingExecutor;
+ this.deviceActionFactory = deviceActionFactory;
this.sharedSchemaRepository = schemaRepositoryProvider.getSharedSchemaRepository();
this.dataBroker = dataBroker;
this.mountPointService = mountPointService;
protected ListenableFuture<NetconfDeviceCapabilities> setupConnection(final NodeId nodeId,
final Node configNode) {
- final NetconfNode netconfNode = configNode.getAugmentation(NetconfNode.class);
+ final NetconfNode netconfNode = configNode.augmentation(NetconfNode.class);
Preconditions.checkNotNull(netconfNode.getHost());
Preconditions.checkNotNull(netconfNode.getPort());
Futures.addCallback(future, new FutureCallback<NetconfDeviceCapabilities>() {
@Override
public void onSuccess(final NetconfDeviceCapabilities result) {
- LOG.debug("Connector for : " + nodeId.getValue() + " started succesfully");
+ LOG.debug("Connector for {} started succesfully", nodeId.getValue());
}
@Override
public void onFailure(final Throwable throwable) {
- LOG.error("Connector for : " + nodeId.getValue() + " failed");
+ LOG.error("Connector for {} failed", nodeId.getValue(), throwable);
// remove this node from active connectors?
}
}, MoreExecutors.directExecutor());
return future;
}
- protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId,
- final NetconfNode node) {
+ protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId, final NetconfNode node) {
//setup default values since default value is not supported in mdsal
final long defaultRequestTimeoutMillis = node.getDefaultRequestTimeoutMillis() == null
? DEFAULT_REQUEST_TIMEOUT_MILLIS : node.getDefaultRequestTimeoutMillis();
if (keepaliveDelay > 0) {
LOG.warn("Adding keepalive facade, for device {}", nodeId);
- salFacade = new KeepaliveSalFacade(remoteDeviceId, salFacade, keepaliveExecutor.getExecutor(),
+ salFacade = new KeepaliveSalFacade(remoteDeviceId, salFacade, this.keepaliveExecutor.getExecutor(),
keepaliveDelay, defaultRequestTimeoutMillis);
}
if (node.isSchemaless()) {
device = new SchemalessNetconfDevice(remoteDeviceId, salFacade);
} else {
- device = new NetconfDeviceBuilder()
+ NetconfDeviceBuilder netconfDeviceBuilder = new NetconfDeviceBuilder()
.setReconnectOnSchemasChange(reconnectOnChangedSchema)
.setSchemaResourcesDTO(schemaResourcesDTO)
- .setGlobalProcessingExecutor(processingExecutor.getExecutor())
+ .setGlobalProcessingExecutor(this.processingExecutor.getExecutor())
.setId(remoteDeviceId)
- .setSalFacade(salFacade)
- .build();
+ .setSalFacade(salFacade);
+ if (this.deviceActionFactory != null) {
+ netconfDeviceBuilder.setDeviceActionFactory(this.deviceActionFactory);
+ }
+ device = netconfDeviceBuilder.build();
}
final Optional<UserPreferences> userCapabilities = getUserCapabilities(node);
LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", remoteDeviceId);
}
- return new NetconfConnectorDTO(userCapabilities.isPresent()
- ? new NetconfDeviceCommunicator(remoteDeviceId, device, userCapabilities.get(), rpcMessageLimit)
- : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit), salFacade);
+ NetconfDeviceCommunicator netconfDeviceCommunicator =
+ userCapabilities.isPresent() ? new NetconfDeviceCommunicator(remoteDeviceId, device,
+ userCapabilities.get(), rpcMessageLimit)
+ : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit);
+
+ if (salFacade instanceof KeepaliveSalFacade) {
+ ((KeepaliveSalFacade)salFacade).setListener(netconfDeviceCommunicator);
+ }
+ return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade);
}
protected NetconfDevice.SchemaResourcesDTO setupSchemaCacheDTO(final NodeId nodeId, final NetconfNode node) {
final FilesystemSchemaSourceCache<YangTextSchemaSource> deviceCache =
createDeviceFilesystemCache(moduleSchemaCacheDirectory);
repository.registerSchemaSourceListener(deviceCache);
+ repository.registerSchemaSourceListener(
+ InMemorySchemaSourceCache.createSoftCache(repository, ASTSchemaSource.class));
return new NetconfDevice.SchemaResourcesDTO(repository, repository, contextFactory,
new NetconfStateSchemasResolverImpl());
}
protected abstract RemoteDeviceHandler<NetconfSessionPreferences> createSalFacade(RemoteDeviceId id);
- private InetSocketAddress getSocketAddress(final Host host, final int port) {
+ private static InetSocketAddress getSocketAddress(final Host host, final int port) {
if (host.getDomainName() != null) {
return new InetSocketAddress(host.getDomainName().getValue(), port);
- } else {
- final IpAddress ipAddress = host.getIpAddress();
- final String ip = ipAddress.getIpv4Address() != null
- ? ipAddress.getIpv4Address().getValue() : ipAddress.getIpv6Address().getValue();
- return new InetSocketAddress(ip, port);
}
+
+ final IpAddress ipAddress = host.getIpAddress();
+ final String ip = ipAddress.getIpv4Address() != null ? ipAddress.getIpv4Address().getValue()
+ : ipAddress.getIpv6Address().getValue();
+ return new InetSocketAddress(ip, port);
}
- private Optional<UserPreferences> getUserCapabilities(final NetconfNode node) {
+ private static Optional<UserPreferences> getUserCapabilities(final NetconfNode node) {
// if none of yang-module-capabilities or non-module-capabilities is specified
// just return absent
if (node.getYangModuleCapabilities() == null && node.getNonModuleCapabilities() == null) {
Code Review / netconf.git / blobdiff