X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-netconf-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fconnect%2Fnetconf%2FNetconfDevice.xtend;h=21500e1da6441540344daf741dd6becec70ed5a6;hp=bfe352ad41322cf78404426a5582afc22a4e074d;hb=ae412f5decbfe52480d70407568fbb8f9c43fecb;hpb=4043d42c401e0ad6369c9ec35f2c926dcc18c80d diff --git a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend index bfe352ad41..21500e1da6 100644 --- a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend +++ b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend @@ -1,61 +1,55 @@ package org.opendaylight.controller.sal.connect.netconf -import org.opendaylight.controller.sal.core.api.mount.MountProvisionInstance -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier -import org.opendaylight.controller.md.sal.common.api.data.DataReader -import org.opendaylight.yangtools.yang.data.api.CompositeNode -import org.opendaylight.controller.netconf.client.NetconfClient -import org.opendaylight.controller.sal.core.api.RpcImplementation -import static extension org.opendaylight.controller.sal.connect.netconf.NetconfMapping.* +import com.google.common.base.Optional +import com.google.common.collect.FluentIterable +import io.netty.util.concurrent.EventExecutor +import java.io.InputStream import java.net.InetSocketAddress -import org.opendaylight.yangtools.yang.data.api.Node -import org.opendaylight.yangtools.yang.data.api.SimpleNode -import org.opendaylight.yangtools.yang.common.QName +import java.net.URI import java.util.Collections +import java.util.List +import java.util.Set +import java.util.concurrent.ExecutorService +import java.util.concurrent.Future +import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler +import org.opendaylight.controller.md.sal.common.api.data.DataModification +import org.opendaylight.controller.md.sal.common.api.data.DataReader +import org.opendaylight.controller.netconf.api.NetconfMessage +import org.opendaylight.controller.netconf.client.NetconfClient import org.opendaylight.controller.netconf.client.NetconfClientDispatcher -import org.opendaylight.yangtools.concepts.Registration -import org.opendaylight.controller.sal.core.api.Provider import org.opendaylight.controller.sal.core.api.Broker.ProviderSession -import org.opendaylight.controller.sal.core.api.mount.MountProvisionService -import static org.opendaylight.controller.sal.connect.netconf.InventoryUtils.*; +import org.opendaylight.controller.sal.core.api.Provider +import org.opendaylight.controller.sal.core.api.RpcImplementation import org.opendaylight.controller.sal.core.api.data.DataBrokerService import org.opendaylight.controller.sal.core.api.data.DataModificationTransaction -import org.opendaylight.yangtools.yang.data.impl.SimpleNodeTOImpl -import org.opendaylight.yangtools.yang.data.impl.CompositeNodeTOImpl +import org.opendaylight.controller.sal.core.api.mount.MountProvisionInstance +import org.opendaylight.controller.sal.core.api.mount.MountProvisionService import org.opendaylight.protocol.framework.ReconnectStrategy -import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler -import org.opendaylight.controller.md.sal.common.api.data.DataModification -import com.google.common.collect.FluentIterable -import org.opendaylight.yangtools.yang.model.api.SchemaContext -import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.NetconfState +import org.opendaylight.yangtools.concepts.Registration +import org.opendaylight.yangtools.yang.common.QName +import org.opendaylight.yangtools.yang.data.api.CompositeNode +import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier +import org.opendaylight.yangtools.yang.data.api.Node +import org.opendaylight.yangtools.yang.data.api.SimpleNode +import org.opendaylight.yangtools.yang.data.impl.CompositeNodeTOImpl +import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode +import org.opendaylight.yangtools.yang.data.impl.SimpleNodeTOImpl +import org.opendaylight.yangtools.yang.model.api.SchemaContext +import org.opendaylight.yangtools.yang.model.util.repo.AbstractCachingSchemaSourceProvider +import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProvider +import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProviders +import org.opendaylight.yangtools.yang.model.util.repo.SourceIdentifier import org.opendaylight.yangtools.yang.parser.impl.YangParserImpl -import java.io.InputStream -import org.slf4j.LoggerFactory +import org.opendaylight.yangtools.yang.parser.impl.util.YangSourceContext import org.slf4j.Logger -import org.opendaylight.controller.netconf.client.AbstractNetconfClientNotifySessionListener -import org.opendaylight.controller.netconf.client.NetconfClientSession -import org.opendaylight.controller.netconf.api.NetconfMessage -import io.netty.util.concurrent.EventExecutor +import org.slf4j.LoggerFactory -import java.util.Map -import java.util.Set -import com.google.common.collect.ImmutableMap +import static com.google.common.base.Preconditions.* +import static org.opendaylight.controller.sal.connect.netconf.InventoryUtils.* -import org.opendaylight.yangtools.yang.model.util.repo.AbstractCachingSchemaSourceProvider -import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProvider -import com.google.common.base.Optional -import com.google.common.collect.ImmutableList -import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProviders -import static com.google.common.base.Preconditions.*; -import java.util.concurrent.ExecutorService -import java.util.concurrent.Future -import org.opendaylight.controller.netconf.client.NetconfClientSessionListener -import io.netty.util.concurrent.Promise -import org.opendaylight.controller.netconf.util.xml.XmlElement -import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants -import java.util.concurrent.ExecutionException -import java.util.concurrent.locks.ReentrantLock +import static extension org.opendaylight.controller.sal.connect.netconf.NetconfMapping.* +import org.opendaylight.controller.netconf.util.xml.XmlUtil class NetconfDevice implements Provider, // DataReader, // @@ -86,7 +80,8 @@ AutoCloseable { @Property var AbstractCachingSchemaSourceProvider schemaSourceProvider; - private NetconfDeviceSchemaContextProvider schemaContextProvider + @Property + private NetconfDeviceSchemaContextProvider deviceContextProvider protected val Logger logger @@ -105,9 +100,12 @@ AutoCloseable { @Property var NetconfClientDispatcher dispatcher - + static val InstanceIdentifier ROOT_PATH = InstanceIdentifier.builder().toInstance(); + @Property + var SchemaSourceProvider remoteSourceProvider + public new(String name) { this.name = name; this.logger = LoggerFactory.getLogger(NetconfDevice.name + "#" + name); @@ -120,11 +118,12 @@ AutoCloseable { checkState(schemaSourceProvider != null, "Schema Source Provider must be set.") checkState(eventExecutor != null, "Event executor must be set."); - val listener = new NetconfDeviceListener(this,eventExecutor); + val listener = new NetconfDeviceListener(this, eventExecutor); val task = startClientTask(dispatcher, listener) - if(mountInstance != null) { + if (mountInstance != null) { confReaderReg = mountInstance.registerConfigurationReader(ROOT_PATH, this); operReaderReg = mountInstance.registerOperationalReader(ROOT_PATH, this); + commitHandlerReg = mountInstance.registerCommitHandler(ROOT_PATH, this) } return processingExecutor.submit(task) as Future; @@ -132,27 +131,30 @@ AutoCloseable { } def Optional getSchemaContext() { - if (schemaContextProvider == null) { + if (deviceContextProvider == null) { return Optional.absent(); } - return schemaContextProvider.currentContext; + return deviceContextProvider.currentContext; } private def Runnable startClientTask(NetconfClientDispatcher dispatcher, NetconfDeviceListener listener) { + return [ | logger.info("Starting Netconf Client on: {}", socketAddress); client = NetconfClient.clientFor(name, socketAddress, reconnectStrategy, dispatcher, listener); logger.debug("Initial capabilities {}", initialCapabilities); var SchemaSourceProvider delegate; - if (initialCapabilities.contains(NetconfMapping.IETF_NETCONF_MONITORING_MODULE)) { - delegate = new NetconfDeviceSchemaSourceProvider(this); + if (NetconfRemoteSchemaSourceProvider.isSupportedFor(initialCapabilities)) { + delegate = new NetconfRemoteSchemaSourceProvider(this); + } else if(client.capabilities.contains(NetconfRemoteSchemaSourceProvider.IETF_NETCONF_MONITORING.namespace.toString)) { + delegate = new NetconfRemoteSchemaSourceProvider(this); } else { - logger.info("Device does not support IETF Netconf Monitoring.", socketAddress); + logger.info("Netconf server {} does not support IETF Netconf Monitoring", socketAddress); delegate = SchemaSourceProviders.noopProvider(); } - val sourceProvider = schemaSourceProvider.createInstanceFor(delegate); - schemaContextProvider = new NetconfDeviceSchemaContextProvider(this, sourceProvider); - schemaContextProvider.createContextFromCapabilities(initialCapabilities); + remoteSourceProvider = schemaSourceProvider.createInstanceFor(delegate); + deviceContextProvider = new NetconfDeviceSchemaContextProvider(this, remoteSourceProvider); + deviceContextProvider.createContextFromCapabilities(initialCapabilities); if (mountInstance != null && schemaContext.isPresent) { mountInstance.schemaContext = schemaContext.get(); } @@ -175,18 +177,31 @@ AutoCloseable { override getSupportedRpcs() { Collections.emptySet; } - + def createSubscription(String streamName) { val it = ImmutableCompositeNode.builder() QName = NETCONF_CREATE_SUBSCRIPTION_QNAME - addLeaf("stream",streamName); - invokeRpc(QName,toInstance()) + addLeaf("stream", streamName); + invokeRpc(QName, toInstance()) } override invokeRpc(QName rpc, CompositeNode input) { - val message = rpc.toRpcMessage(input); - val result = client.sendMessage(message, messegeRetryCount, messageTimeoutCount); - return result.toRpcResult(); + try { + val message = rpc.toRpcMessage(input,schemaContext); + val result = sendMessageImpl(message, messegeRetryCount, messageTimeoutCount); + return result.toRpcResult(rpc, schemaContext); + + } catch (Exception e) { + logger.error("Rpc was not processed correctly.", e) + throw e; + } + } + + def NetconfMessage sendMessageImpl(NetconfMessage message, int retryCount, int timeout) { + logger.debug("Send message {}",XmlUtil.toString(message.document)) + val result = client.sendMessage(message, retryCount, timeout); + NetconfMapping.checkValidReply(message, result) + return result; } override getProviderFunctionality() { @@ -221,7 +236,7 @@ AutoCloseable { return null === transaction.readOperationalData(path); } - def Node findNode(CompositeNode node, InstanceIdentifier identifier) { + static def Node findNode(CompositeNode node, InstanceIdentifier identifier) { var Node current = node; for (arg : identifier.path) { @@ -229,12 +244,17 @@ AutoCloseable { return null; } else if (current instanceof CompositeNode) { val currentComposite = (current as CompositeNode); - - current = currentComposite.getFirstCompositeByName(arg.nodeType.withoutRevision()); - if (current == null) { - current = currentComposite.getFirstSimpleByName(arg.nodeType.withoutRevision()); + + current = currentComposite.getFirstCompositeByName(arg.nodeType); + if(current == null) { + current = currentComposite.getFirstCompositeByName(arg.nodeType.withoutRevision()); + } + if(current == null) { + current = currentComposite.getFirstSimpleByName(arg.nodeType); } if (current == null) { + current = currentComposite.getFirstSimpleByName(arg.nodeType.withoutRevision()); + } if (current == null) { return null; } } @@ -243,7 +263,9 @@ AutoCloseable { } override requestCommit(DataModification modification) { - throw new UnsupportedOperationException("TODO: auto-generated method stub") + val twoPhaseCommit = new NetconfDeviceTwoPhaseCommitTransaction(this, modification); + twoPhaseCommit.prepare() + return twoPhaseCommit; } def getInitialCapabilities() { @@ -257,8 +279,18 @@ AutoCloseable { val parts = split("\\?"); val namespace = parts.get(0); val queryParams = FluentIterable.from(parts.get(1).split("&")); - val revision = queryParams.findFirst[startsWith("revision=")].replaceAll("revision=", ""); - val moduleName = queryParams.findFirst[startsWith("module=")].replaceAll("module=", ""); + var revision = queryParams.findFirst[startsWith("revision=")]?.replaceAll("revision=", ""); + val moduleName = queryParams.findFirst[startsWith("module=")]?.replaceAll("module=", ""); + if (revision === null) { + logger.warn("Netconf device was not reporting revision correctly, trying to get amp;revision="); + revision = queryParams.findFirst[startsWith("&revision=")]?.replaceAll("revision=", ""); + if (revision != null) { + logger.warn("Netconf device returned revision incorectly escaped for {}", it) + } + } + if (revision == null) { + return QName.create(URI.create(namespace), null, moduleName); + } return QName.create(namespace, revision, moduleName); ].toSet(); } @@ -273,96 +305,6 @@ AutoCloseable { } -package class NetconfDeviceListener extends NetconfClientSessionListener { - - val NetconfDevice device - val EventExecutor eventExecutor - - new(NetconfDevice device,EventExecutor eventExecutor) { - this.device = device - this.eventExecutor = eventExecutor - } - - var Promise messagePromise; - val promiseLock = new ReentrantLock; - - override onMessage(NetconfClientSession session, NetconfMessage message) { - if (isNotification(message)) { - onNotification(session, message); - } else try { - promiseLock.lock - if (messagePromise != null) { - messagePromise.setSuccess(message); - messagePromise = null; - } - } finally { - promiseLock.unlock - } - } - - /** - * Method intended to customize notification processing. - * - * @param session - * {@see - * NetconfClientSessionListener#onMessage(NetconfClientSession, - * NetconfMessage)} - * @param message - * {@see - * NetconfClientSessionListener#onMessage(NetconfClientSession, - * NetconfMessage)} - */ - def void onNotification(NetconfClientSession session, NetconfMessage message) { - device.logger.debug("Received NETCONF notification.",message); - val domNotification = message?.toCompositeNode?.notificationBody; - if(domNotification != null) { - device?.mountInstance?.publish(domNotification); - } - } - - private static def CompositeNode getNotificationBody(CompositeNode node) { - for(child : node.children) { - if(child instanceof CompositeNode) { - return child as CompositeNode; - } - } - } - - override getLastMessage(int attempts, int attemptMsDelay) throws InterruptedException { - val promise = promiseReply(); - val messageAvailable = promise.await(attempts + attemptMsDelay); - if (messageAvailable) { - try { - return promise.get(); - } catch (ExecutionException e) { - throw new IllegalStateException(e); - } - } - - throw new IllegalStateException("Unsuccessful after " + attempts + " attempts."); - - // throw new TimeoutException("Message was not received on time."); - } - - def Promise promiseReply() { - promiseLock.lock - try { - if (messagePromise == null) { - messagePromise = eventExecutor.newPromise(); - return messagePromise; - } - return messagePromise; - } finally { - promiseLock.unlock - } - } - - def boolean isNotification(NetconfMessage message) { - val xmle = XmlElement.fromDomDocument(message.getDocument()); - return XmlNetconfConstants.NOTIFICATION_ELEMENT_NAME.equals(xmle.getName()); - } -} - package class NetconfDeviceSchemaContextProvider { @Property @@ -377,25 +319,33 @@ package class NetconfDeviceSchemaContextProvider { new(NetconfDevice device, SchemaSourceProvider sourceProvider) { _device = device _sourceProvider = sourceProvider + _currentContext = Optional.absent(); } def createContextFromCapabilities(Iterable capabilities) { - - val modelsToParse = ImmutableMap.builder(); - for (cap : capabilities) { - val source = sourceProvider.getSchemaSource(cap.localName, Optional.fromNullable(cap.formattedRevision)); - if (source.present) { - modelsToParse.put(cap, source.get()); - } + val sourceContext = YangSourceContext.createFrom(capabilities, sourceProvider) + if (!sourceContext.missingSources.empty) { + device.logger.warn("Sources for following models are missing {}", sourceContext.missingSources); + } + device.logger.debug("Trying to create schema context from {}", sourceContext.validSources) + val modelsToParse = YangSourceContext.getValidInputStreams(sourceContext); + if (!sourceContext.validSources.empty) { + val schemaContext = tryToCreateContext(modelsToParse); + currentContext = Optional.fromNullable(schemaContext); + } else { + currentContext = Optional.absent(); } - val context = tryToCreateContext(modelsToParse.build); - currentContext = Optional.fromNullable(context); + if (currentContext.present) { + device.logger.debug("Schema context successfully created."); + } + } - def SchemaContext tryToCreateContext(Map modelsToParse) { + def SchemaContext tryToCreateContext(List modelsToParse) { val parser = new YangParserImpl(); try { - val models = parser.parseYangModelsFromStreams(ImmutableList.copyOf(modelsToParse.values)); + + val models = parser.parseYangModelsFromStreams(modelsToParse); val result = parser.resolveSchemaContext(models); return result; } catch (Exception e) { @@ -404,33 +354,3 @@ package class NetconfDeviceSchemaContextProvider { } } } - -package class NetconfDeviceSchemaSourceProvider implements SchemaSourceProvider { - - val NetconfDevice device; - - new(NetconfDevice device) { - this.device = device; - } - - override getSchemaSource(String moduleName, Optional revision) { - val it = ImmutableCompositeNode.builder() // - setQName(QName::create(NetconfState.QNAME, "get-schema")) // - addLeaf("format", "yang") - addLeaf("identifier", moduleName) - if (revision.present) { - addLeaf("version", revision.get()) - } - - device.logger.info("Loading YANG schema source for {}:{}", moduleName, revision) - val schemaReply = device.invokeRpc(getQName(), toInstance()); - - if (schemaReply.successful) { - val schemaBody = schemaReply.result.getFirstSimpleByName( - QName::create(NetconfState.QNAME.namespace, null, "data"))?.value; - device.logger.info("YANG Schema successfully received for: {}:{}", moduleName, revision); - return Optional.of(schemaBody as String); - } - return Optional.absent(); - } -}