X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=netconf%2Fnetconf-topology%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fnetconf%2Ftopology%2FAbstractNetconfTopology.java;h=66eea5692b4838f2fec4d81e167135de3f3087f1;hb=refs%2Fchanges%2F97%2F77797%2F1;hp=9bc6b59e7e8ccc7788df3144a801c11ee9bfa4a2;hpb=32c6b3e6afb7d118d5615ec001a9f53705b8eecb;p=netconf.git
diff --git a/netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/AbstractNetconfTopology.java b/netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/AbstractNetconfTopology.java
index 9bc6b59e7e..66eea5692b 100644
--- a/netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/AbstractNetconfTopology.java
+++ b/netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/AbstractNetconfTopology.java
@@ -16,7 +16,9 @@ import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
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;
@@ -31,6 +33,7 @@ import java.util.HashMap;
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;
@@ -49,6 +52,7 @@ import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurati
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;
@@ -95,7 +99,9 @@ import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
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;
@@ -141,12 +147,8 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
private static final SharedSchemaRepository DEFAULT_SCHEMA_REPOSITORY =
new SharedSchemaRepository(DEFAULT_SCHEMA_REPOSITORY_NAME);
- /**
- * The default FilesystemSchemaSourceCache
, which stores cached files in cache/schema
.
- */
- private static final FilesystemSchemaSourceCache DEFAULT_CACHE =
- new FilesystemSchemaSourceCache<>(DEFAULT_SCHEMA_REPOSITORY, YangTextSchemaSource.class,
- new File(QUALIFIED_DEFAULT_CACHE_DIRECTORY));
+ public static final InMemorySchemaSourceCache DEFAULT_AST_CACHE =
+ InMemorySchemaSourceCache.createSoftCache(DEFAULT_SCHEMA_REPOSITORY, ASTSchemaSource.class);
/**
* The default factory for creating SchemaContext
instances.
@@ -171,20 +173,45 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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 FilesystemSchemaSourceCache
, which stores cached files
+ * in cache/schema
. 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 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 ListeningExecutorService 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;
@@ -198,12 +225,14 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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.processingExecutor = MoreExecutors.listeningDecorator(processingExecutor.getExecutor());
+ this.deviceActionFactory = deviceActionFactory;
this.sharedSchemaRepository = schemaRepositoryProvider.getSharedSchemaRepository();
this.dataBroker = dataBroker;
this.mountPointService = mountPointService;
@@ -243,7 +272,7 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
protected ListenableFuture 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());
@@ -262,12 +291,12 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
Futures.addCallback(future, new FutureCallback() {
@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());
@@ -275,14 +304,13 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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
+ final long defaultRequestTimeoutMillis = node.getDefaultRequestTimeoutMillis() == null
? DEFAULT_REQUEST_TIMEOUT_MILLIS : node.getDefaultRequestTimeoutMillis();
- final Long keepaliveDelay = node.getKeepaliveDelay() == null
+ final long keepaliveDelay = node.getKeepaliveDelay() == null
? DEFAULT_KEEPALIVE_DELAY : node.getKeepaliveDelay();
- final Boolean reconnectOnChangedSchema = node.isReconnectOnChangedSchema() == null
+ final boolean reconnectOnChangedSchema = node.isReconnectOnChangedSchema() == null
? DEFAULT_RECONNECT_ON_CHANGED_SCHEMA : node.isReconnectOnChangedSchema();
final IpAddress ipAddress = node.getHost().getIpAddress();
@@ -296,7 +324,7 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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);
}
@@ -332,13 +360,16 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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)
.setId(remoteDeviceId)
- .setSalFacade(salFacade)
- .build();
+ .setSalFacade(salFacade);
+ if (this.deviceActionFactory != null) {
+ netconfDeviceBuilder.setDeviceActionFactory(this.deviceActionFactory);
+ }
+ device = netconfDeviceBuilder.build();
}
final Optional userCapabilities = getUserCapabilities(node);
@@ -349,9 +380,15 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
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) {
@@ -410,6 +447,8 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
final FilesystemSchemaSourceCache deviceCache =
createDeviceFilesystemCache(moduleSchemaCacheDirectory);
repository.registerSchemaSourceListener(deviceCache);
+ repository.registerSchemaSourceListener(
+ InMemorySchemaSourceCache.createSoftCache(repository, ASTSchemaSource.class));
return new NetconfDevice.SchemaResourcesDTO(repository, repository, contextFactory,
new NetconfStateSchemasResolverImpl());
}
@@ -477,6 +516,10 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
} else {
throw new IllegalStateException("Unsupported protocol type: " + node.getProtocol().getName().getClass());
}
+ if (node.getOdlHelloMessageCapabilities() != null) {
+ reconnectingClientConfigurationBuilder
+ .withOdlHelloCapabilities(node.getOdlHelloMessageCapabilities().getCapability());
+ }
return reconnectingClientConfigurationBuilder
.withAddress(socketAddress)
@@ -516,18 +559,18 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
protected abstract RemoteDeviceHandler 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 getUserCapabilities(final NetconfNode node) {
+ private static Optional 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) {
@@ -579,11 +622,8 @@ public abstract class AbstractNetconfTopology implements NetconfTopology {
@Override
public ReconnectStrategy createReconnectStrategy() {
- final Long maxSleep = null;
- final Long deadline = null;
-
return new TimedReconnectStrategy(executor, minSleep,
- minSleep, sleepFactor, maxSleep, connectionAttempts, deadline);
+ minSleep, sleepFactor, null /*maxSleep*/, connectionAttempts, null /*deadline*/);
}
}