From: Tony Tkacik Date: Mon, 1 Dec 2014 10:18:09 +0000 (+0000) Subject: Merge "Fix checkstyle warnings in netconf-tcp" X-Git-Tag: release/lithium~823 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=commitdiff_plain;h=44eb0b8e2fc5f240cfca167207e026d35b75f8a5;hp=f2d3fedde9aa2af532096fe726b6522155d873ac;p=controller.git Merge "Fix checkstyle warnings in netconf-tcp" --- diff --git a/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.java b/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.java index 98a2c2cca1..b2ab27a826 100644 --- a/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.java +++ b/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.java @@ -80,7 +80,7 @@ final class ReconnectPromise, L extends SessionList /** * Channel handler that responds to channelInactive event and reconnects the session. - * Only if the initial connection was successfully established and promise was not canceled. + * Only if the promise was not canceled. */ private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter { private final ReconnectPromise promise; diff --git a/opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/ServerTest.java b/opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/ServerTest.java index fc38888de3..6c4af0186f 100644 --- a/opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/ServerTest.java +++ b/opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/ServerTest.java @@ -250,6 +250,52 @@ public class ServerTest { assertFalse(session.isSuccess()); } + @Test + public void testNegotiationFailedReconnect() throws Exception { + final Promise p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE); + + this.dispatcher = getServerDispatcher(p); + + this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory() { + @Override + public SimpleSessionListener getSessionListener() { + return new SimpleSessionListener(); + } + }); + + this.server.get(); + + this.clientDispatcher = new SimpleDispatcher(new SessionNegotiatorFactory() { + @Override + public SessionNegotiator getSessionNegotiator(final SessionListenerFactory factory, + final Channel channel, final Promise promise) { + + return new SimpleSessionNegotiator(promise, channel) { + @Override + protected void startNegotiation() throws Exception { + negotiationFailed(new IllegalStateException("Negotiation failed")); + } + }; + } + }, new DefaultPromise(GlobalEventExecutor.INSTANCE), eventLoopGroup); + + final ReconnectStrategyFactory reconnectStrategyFactory = mock(ReconnectStrategyFactory.class); + final ReconnectStrategy reconnectStrategy = getMockedReconnectStrategy(); + doReturn(reconnectStrategy).when(reconnectStrategyFactory).createReconnectStrategy(); + + this.clientDispatcher.createReconnectingClient(this.serverAddress, + reconnectStrategyFactory, new SessionListenerFactory() { + @Override + public SimpleSessionListener getSessionListener() { + return new SimpleSessionListener(); + } + }); + + + // Reconnect strategy should be consulted at least twice, for initial connect and reconnect attempts after drop + verify(reconnectStrategyFactory, timeout((int) TimeUnit.MINUTES.toMillis(3)).atLeast(2)).createReconnectStrategy(); + } + private SimpleDispatcher getClientDispatcher() { return new SimpleDispatcher(new SessionNegotiatorFactory() { @Override diff --git a/opendaylight/md-sal/pom.xml b/opendaylight/md-sal/pom.xml index 1f646f2bba..38756f2906 100644 --- a/opendaylight/md-sal/pom.xml +++ b/opendaylight/md-sal/pom.xml @@ -141,6 +141,19 @@ + + + org.opendaylight.yangtools + yang-maven-plugin + [0.5,) + + generate-sources + + + + + + @@ -227,4 +240,4 @@ - \ No newline at end of file + diff --git a/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareConsumer.java b/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareConsumer.java index 02bb6b35b9..fc9d9fa2ba 100644 --- a/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareConsumer.java +++ b/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareConsumer.java @@ -9,6 +9,7 @@ package org.opendaylight.controller.sal.binding.api; import org.osgi.framework.BundleContext; +@Deprecated public abstract class AbstractBindingAwareConsumer extends AbstractBrokerAwareActivator implements BindingAwareConsumer { @Override diff --git a/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareProvider.java b/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareProvider.java index 068b6c204f..d32fa36b66 100644 --- a/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareProvider.java +++ b/opendaylight/md-sal/sal-binding-api/src/main/java/org/opendaylight/controller/sal/binding/api/AbstractBindingAwareProvider.java @@ -15,6 +15,7 @@ import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderCo import org.opendaylight.yangtools.yang.binding.RpcService; import org.osgi.framework.BundleContext; +@Deprecated public abstract class AbstractBindingAwareProvider extends AbstractBrokerAwareActivator implements BindingAwareProvider { @Override diff --git a/opendaylight/md-sal/sal-dom-xsql/src/main/java/org/opendaylight/controller/md/sal/dom/xsql/XSQLBluePrintRelation.java b/opendaylight/md-sal/sal-dom-xsql/src/main/java/org/opendaylight/controller/md/sal/dom/xsql/XSQLBluePrintRelation.java index 9053f00aae..55a35e041a 100644 --- a/opendaylight/md-sal/sal-dom-xsql/src/main/java/org/opendaylight/controller/md/sal/dom/xsql/XSQLBluePrintRelation.java +++ b/opendaylight/md-sal/sal-dom-xsql/src/main/java/org/opendaylight/controller/md/sal/dom/xsql/XSQLBluePrintRelation.java @@ -81,59 +81,39 @@ public class XSQLBluePrintRelation implements Serializable { } public List execute(Object o) { - List result = new LinkedList<>(); if (o == null) { return null; } - if (Set.class.isAssignableFrom(o.getClass())) { - Set lst = (Set) o; - for (Object oo : lst) { + List result = new LinkedList<>(); + if (o instanceof Set) { + for (Object oo : (Set) o) { addToResult(result, execute(oo)); } - return result; - } else if (List.class.isAssignableFrom(o.getClass())) { - List lst = (List) o; - for (Object oo : lst) { + } else if (o instanceof List) { + for (Object oo : (List) o) { addToResult(result, execute(oo)); } - return result; - } else if (Map.class.isAssignableFrom(o.getClass())) { - Map map = (Map) o; - for (Object oo : map.values()) { + } else if (o instanceof Map) { + for (Object oo : ((Map) o).values()) { addToResult(result, execute(oo)); } - return result; + } else { + addToResult(result, XSQLCriteria.getValue(o, this.property)); } - - addToResult(result, XSQLCriteria.getValue(o, this.property)); - return result; } private static void addToResult(List result, Object o) { - if (o == null) { - return; - } - if (Set.class.isAssignableFrom(o.getClass())) { - Set lst = (Set) o; - for (Object oo : lst) { - result.add(oo); - } - } else if (List.class.isAssignableFrom(o.getClass())) { - List lst = (List) o; - for (Object oo : lst) { - result.add(oo); - } - } else if (Map.class.isAssignableFrom(o.getClass())) { - Map map = (Map) o; - for (Object oo : map.values()) { - result.add(oo); - } - } else { + if (o instanceof Set) { + result.addAll((Set)o); + } else if (o instanceof List) { + result.addAll((List)o); + } else if (o instanceof Map) { + result.addAll(((Map)o).values()); + } else if (o != null) { result.add(o); } } - } diff --git a/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicatorTest.java b/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicatorTest.java index 001b9a8d3a..a24034d2f0 100644 --- a/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicatorTest.java +++ b/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicatorTest.java @@ -19,6 +19,8 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.RPC_REPLY_KEY; import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0; @@ -27,9 +29,15 @@ import com.google.common.base.Strings; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ListenableFuture; import io.netty.channel.ChannelFuture; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; +import io.netty.util.concurrent.GlobalEventExecutor; import java.io.ByteArrayInputStream; +import java.net.InetSocketAddress; import java.util.Collection; import java.util.Collections; import java.util.UUID; @@ -45,11 +53,18 @@ import org.mockito.Mock; import org.mockito.MockitoAnnotations; import org.opendaylight.controller.netconf.api.NetconfMessage; import org.opendaylight.controller.netconf.api.NetconfTerminationReason; +import org.opendaylight.controller.netconf.client.NetconfClientDispatcherImpl; import org.opendaylight.controller.netconf.client.NetconfClientSession; +import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration; +import org.opendaylight.controller.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.LoginPassword; import org.opendaylight.controller.sal.connect.api.RemoteDevice; import org.opendaylight.controller.sal.connect.api.RemoteDeviceCommunicator; import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil; import org.opendaylight.controller.sal.connect.util.RemoteDeviceId; +import org.opendaylight.protocol.framework.ReconnectStrategy; +import org.opendaylight.protocol.framework.ReconnectStrategyFactory; +import org.opendaylight.protocol.framework.TimedReconnectStrategy; import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; @@ -320,6 +335,60 @@ public class NetconfDeviceCommunicatorTest { errorInfo.contains( "bar" ) ); } + /** + * Test whether reconnect is scheduled properly + */ + @Test + public void testNetconfDeviceReconnectInCommunicator() throws Exception { + final RemoteDevice device = mock(RemoteDevice.class); + + final TimedReconnectStrategy timedReconnectStrategy = new TimedReconnectStrategy(GlobalEventExecutor.INSTANCE, 10000, 0, 1.0, null, 100L, null); + final ReconnectStrategy reconnectStrategy = spy(new ReconnectStrategy() { + @Override + public int getConnectTimeout() throws Exception { + return timedReconnectStrategy.getConnectTimeout(); + } + + @Override + public Future scheduleReconnect(final Throwable cause) { + return timedReconnectStrategy.scheduleReconnect(cause); + } + + @Override + public void reconnectSuccessful() { + timedReconnectStrategy.reconnectSuccessful(); + } + }); + + final NetconfDeviceCommunicator listener = new NetconfDeviceCommunicator(new RemoteDeviceId("test"), device); + final EventLoopGroup group = new NioEventLoopGroup(); + final Timer time = new HashedWheelTimer(); + try { + final NetconfClientConfiguration cfg = NetconfReconnectingClientConfigurationBuilder.create() + .withAddress(new InetSocketAddress("localhost", 65000)) + .withReconnectStrategy(reconnectStrategy) + .withConnectStrategyFactory(new ReconnectStrategyFactory() { + @Override + public ReconnectStrategy createReconnectStrategy() { + return reconnectStrategy; + } + }) + .withAuthHandler(new LoginPassword("admin", "admin")) + .withConnectionTimeoutMillis(10000) + .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH) + .withSessionListener(listener) + .build(); + + + listener.initializeRemoteConnection(new NetconfClientDispatcherImpl(group, group, time), cfg); + + verify(reconnectStrategy, timeout((int) TimeUnit.MINUTES.toMillis(3)).times(101)).scheduleReconnect(any(Throwable.class)); + } finally { + time.stop(); + group.shutdownGracefully(); + } + } + @Test public void testOnResponseMessageWithWrongMessageID() throws Exception { setupSession(); diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatPermCollector.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatPermCollector.java index 9116e66737..16ad28dd4f 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatPermCollector.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatPermCollector.java @@ -82,17 +82,27 @@ public interface StatPermCollector extends Runnable, AutoCloseable { /** * All disconnected Nodes need be removed from stat list Nodes + * * @param flowNode * @return true/false if the {@link Node} removed successful */ boolean disconnectedNodeUnregistration(InstanceIdentifier nodeIdent); + /** + * Method add new feature {@link StatCapabTypes} to Node identified by + * nodeIdent -> InstanceIdentifier + * + * @param flowNode + * @return true/false if the {@link StatCapabTypes} add successful + */ + boolean registerAdditionalNodeFeature(InstanceIdentifier nodeIdent, StatCapabTypes statCapab); + /** * Method return true only and only if {@link StatPermCollector} contain * valid node registration in its internal {@link Node} map. * Otherwise return false. * - * @param InstanceIdentifier flowNode + * @param flowNode * @return */ boolean isProvidedFlowNodeActive(InstanceIdentifier nodeIdent); diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsManager.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsManager.java index b53c54e360..7d57067df1 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsManager.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsManager.java @@ -119,6 +119,16 @@ public interface StatisticsManager extends AutoCloseable, TransactionChainListen */ void disconnectedNodeUnregistration(InstanceIdentifier nodeIdent); + /** + * Method wraps {@link StatPermCollector}.registerAdditionalNodeFeature to provide + * possibility to register additional Node Feature {@link StatCapabTypes} for + * statistics collecting. + * + * @param nodeIdent + * @param statCapab + */ + void registerAdditionalNodeFeature(InstanceIdentifier nodeIdent, StatCapabTypes statCapab); + /** * Method provides access to Device RPC methods by wrapped * internal method. In next {@link StatRpcMsgManager} is registered all diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractListenCommit.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractListenCommit.java index 10bfcba678..68197805ab 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractListenCommit.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractListenCommit.java @@ -51,6 +51,7 @@ public abstract class StatAbstractListenCommit listenerRegistration; protected final Map, Map, Integer>> mapNodesForDelete = new ConcurrentHashMap<>(); + protected final Map, Integer> mapNodeFeautureRepeater = new ConcurrentHashMap<>(); private final Class clazz; @@ -127,6 +128,11 @@ public abstract class StatAbstractListenCommit Optional readLatestConfiguration(final InstanceIdentifier path) { if(currentReadTx == null) { currentReadTx = dataBroker.newReadOnlyTransaction(); diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractNotifyCommit.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractNotifyCommit.java index 08871e9980..6bc6a30f8f 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractNotifyCommit.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatAbstractNotifyCommit.java @@ -110,6 +110,29 @@ public abstract class StatAbstractNotifyCommit i return txContainer; } + /** + * Method validate TransactionCacheContainer. It needs to call before every txCacheContainer processing. + * + * @param txCacheContainer + * @return + */ + protected boolean isTransactionCacheContainerValid(final Optional> txCacheContainer) { + if ( ! txCacheContainer.isPresent()) { + LOG.debug("Transaction Cache Container is not presented!"); + return false; + } + if (txCacheContainer.get().getNodeId() == null) { + LOG.debug("Transaction Cache Container {} don't have Node ID!", txCacheContainer.get().getId()); + return false; + } + if (txCacheContainer.get().getNotifications() == null) { + LOG.debug("Transaction Cache Container {} for {} node don't have Notifications!", + txCacheContainer.get().getId(), txCacheContainer.get().getNodeId()); + return false; + } + return true; + } + /** * Wrapping Future object call to {@link org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager} * isExpectedStatistics with 10sec TimeOut. diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitGroup.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitGroup.java index f351132f7f..5185ef0b82 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitGroup.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitGroup.java @@ -16,6 +16,7 @@ import org.opendaylight.controller.md.sal.binding.api.DataBroker; import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; +import org.opendaylight.controller.md.statistics.manager.StatPermCollector.StatCapabTypes; import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager.TransactionCacheContainer; import org.opendaylight.controller.md.statistics.manager.StatisticsManager; import org.opendaylight.controller.md.statistics.manager.StatisticsManager.StatDataStoreOperation; @@ -29,7 +30,9 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111. import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStats; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStatsBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures; +import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeaturesBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatistics; +import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatisticsBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsListener; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.desc.GroupDescBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.features.GroupFeatures; @@ -51,6 +54,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Optional; +import com.google.common.base.Preconditions; /** * statistics-manager @@ -90,31 +94,48 @@ public class StatListenCommitGroup extends StatAbstractListenCommit groupStats = notification.getGroupDescStats() != null - ? new ArrayList<>(notification.getGroupDescStats()) : new ArrayList(10); - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if (txContainer.isPresent()) { - final List cacheNotifs = - txContainer.get().getNotifications(); - for (final TransactionAware notif : cacheNotifs) { - if (notif instanceof GroupDescStatsUpdated) { - groupStats.addAll(((GroupDescStatsUpdated) notif).getGroupDescStats()); - } - } - } - final InstanceIdentifier nodeIdent = InstanceIdentifier - .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { - statGroupDescCommit(groupStats, nodeIdent, tx); + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + /* Validate exist FlowCapableNode */ + final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); + Optional fNode = Optional.absent(); + try { + fNode = tx.read(LogicalDatastoreType.OPERATIONAL,fNodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); + } + if ( ! fNode.isPresent()) { + return; + } + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; + } + /* Prepare List actual Groups and not updated Groups will be removed */ + final List existGroups = fNode.get().getGroup() != null + ? fNode.get().getGroup() : Collections. emptyList(); + final List existGroupKeys = new ArrayList<>(); + for (final Group group : existGroups) { + existGroupKeys.add(group.getKey()); + } + /* GroupDesc processing */ + statGroupDescCommit(txContainer, tx, fNodeIdent, existGroupKeys); + /* Delete all not presented Group Nodes */ + deleteAllNotPresentNode(fNodeIdent, tx, Collections.unmodifiableList(existGroupKeys)); /* Notification for continue collecting statistics */ notifyToCollectNextStatistics(nodeIdent); } @@ -123,39 +144,53 @@ public class StatListenCommitGroup extends StatAbstractListenCommit> txContainer = getTransactionCacheContainer(transId, nodeId); - if ( ! txContainer.isPresent()) { return; } - final InstanceIdentifier nodeIdent = InstanceIdentifier - .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { - notifyToCollectNextStatistics(nodeIdent); - final GroupFeatures stats = new GroupFeaturesBuilder(notification).build(); - final InstanceIdentifier groupFeatureIdent = nodeIdent - .augmentation(NodeGroupFeatures.class).child(GroupFeatures.class); - Optional node = Optional.absent(); - try { - node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; } - if (node.isPresent()) { - tx.put(LogicalDatastoreType.OPERATIONAL, groupFeatureIdent, stats); + + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + final List cacheNotifs = txContainer.get().getNotifications(); + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof GroupFeaturesUpdated)) { + break; + } + final GroupFeatures stats = new GroupFeaturesBuilder((GroupFeaturesUpdated)notif).build(); + final InstanceIdentifier nodeGroupFeatureIdent = + nodeIdent.augmentation(NodeGroupFeatures.class); + final InstanceIdentifier groupFeatureIdent = nodeGroupFeatureIdent + .child(GroupFeatures.class); + Optional node = Optional.absent(); + try { + node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + } + if (node.isPresent()) { + tx.merge(LogicalDatastoreType.OPERATIONAL, nodeGroupFeatureIdent, new NodeGroupFeaturesBuilder().build(), true); + tx.put(LogicalDatastoreType.OPERATIONAL, groupFeatureIdent, stats); + manager.registerAdditionalNodeFeature(nodeIdent, StatCapabTypes.GROUP_STATS); + } } } }); @@ -163,123 +198,140 @@ public class StatListenCommitGroup extends StatAbstractListenCommit groupStats = notification.getGroupStats() != null - ? new ArrayList<>(notification.getGroupStats()) : new ArrayList(10); - Optional notifGroup = Optional.absent(); - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if (txContainer.isPresent()) { - final Optional inputObj = txContainer.get().getConfInput(); - if (inputObj.isPresent() && inputObj.get() instanceof Group) { - notifGroup = Optional. of((Group)inputObj.get()); - } - final List cacheNotifs = - txContainer.get().getNotifications(); - for (final TransactionAware notif : cacheNotifs) { - if (notif instanceof GroupStatisticsUpdated) { - groupStats.addAll(((GroupStatisticsUpdated) notif).getGroupStats()); - } - } - } - final Optional group = notifGroup; - final InstanceIdentifier nodeIdent = InstanceIdentifier - .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { - /* Notification for continue collecting statistics */ - if ( ! group.isPresent()) { + + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + /* Node exist check */ + Optional node = Optional.absent(); + try { + node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + } + if ( ! node.isPresent()) { + return; + } + + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; + } + final List cacheNotifs = txContainer.get().getNotifications(); + + Optional notifGroup = Optional.absent(); + final Optional inputObj = txContainer.get().getConfInput(); + if (inputObj.isPresent() && inputObj.get() instanceof Group) { + notifGroup = Optional. of((Group)inputObj.get()); + } + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof GroupStatisticsUpdated)) { + break; + } + statGroupCommit(((GroupStatisticsUpdated) notif).getGroupStats(), nodeIdent, tx); + } + if (notifGroup.isPresent()) { notifyToCollectNextStatistics(nodeIdent); } - statGroupCommit(groupStats, nodeIdent, group, tx); } }); } private void statGroupCommit(final List groupStats, final InstanceIdentifier nodeIdent, - final Optional group, final ReadWriteTransaction trans) { + final ReadWriteTransaction tx) { + + Preconditions.checkNotNull(groupStats); + Preconditions.checkNotNull(nodeIdent); + Preconditions.checkNotNull(tx); + final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); - for (final GroupStats groupStat : groupStats) { - final GroupStatistics stats = new GroupStatisticsBuilder(groupStat).build(); + for (final GroupStats gStat : groupStats) { + final GroupStatistics stats = new GroupStatisticsBuilder(gStat).build(); - final GroupKey groupKey = new GroupKey(groupStat.getGroupId()); - final InstanceIdentifier gsIdent = fNodeIdent - .child(Group.class,groupKey).augmentation(NodeGroupStatistics.class) - .child(GroupStatistics.class); + final InstanceIdentifier groupIdent = fNodeIdent.child(Group.class, new GroupKey(gStat.getGroupId())); + final InstanceIdentifier nGroupStatIdent =groupIdent + .augmentation(NodeGroupStatistics.class); + final InstanceIdentifier gsIdent = nGroupStatIdent.child(GroupStatistics.class); /* Statistics Writing */ - Optional fNode = Optional.absent(); + Optional group = Optional.absent(); try { - fNode = trans.read(LogicalDatastoreType.OPERATIONAL, fNodeIdent).checkedGet(); + group = tx.read(LogicalDatastoreType.OPERATIONAL, groupIdent).checkedGet(); } catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); + LOG.debug("Read Operational/DS for Group node fail! {}", groupIdent, e); } - if (fNode.isPresent()) { - trans.put(LogicalDatastoreType.OPERATIONAL, gsIdent, stats); + if (group.isPresent()) { + tx.merge(LogicalDatastoreType.OPERATIONAL, nGroupStatIdent, new NodeGroupStatisticsBuilder().build(), true); + tx.put(LogicalDatastoreType.OPERATIONAL, gsIdent, stats); } } } - private void statGroupDescCommit(final List groupStats, final InstanceIdentifier nodeIdent, - final ReadWriteTransaction trans) { - final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); + private void statGroupDescCommit(final Optional> txContainer, final ReadWriteTransaction tx, + final InstanceIdentifier fNodeIdent, final List existGroupKeys) { - final List deviceGroupKeys = new ArrayList<>(); + Preconditions.checkNotNull(existGroupKeys); + Preconditions.checkNotNull(txContainer); + Preconditions.checkNotNull(fNodeIdent); + Preconditions.checkNotNull(tx); - for (final GroupDescStats group : groupStats) { - if (group.getGroupId() != null) { - final GroupBuilder groupBuilder = new GroupBuilder(group); - final GroupKey groupKey = new GroupKey(group.getGroupId()); - final InstanceIdentifier groupRef = fNodeIdent.child(Group.class,groupKey); + final List cacheNotifs = txContainer.get().getNotifications(); + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof GroupDescStatsUpdated)) { + break; + } + final List groupStats = ((GroupDescStatsUpdated) notif).getGroupDescStats(); + if (groupStats == null) { + break; + } + for (final GroupDescStats group : groupStats) { + if (group.getGroupId() != null) { + final GroupBuilder groupBuilder = new GroupBuilder(group); + final GroupKey groupKey = new GroupKey(group.getGroupId()); + final InstanceIdentifier groupRef = fNodeIdent.child(Group.class,groupKey); - final NodeGroupDescStatsBuilder groupDesc= new NodeGroupDescStatsBuilder(); - groupDesc.setGroupDesc(new GroupDescBuilder(group).build()); - //Update augmented data - groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build()); - deviceGroupKeys.add(groupKey); - Optional hashIdUpd = Optional.absent(); - try { - hashIdUpd = trans.read(LogicalDatastoreType.OPERATIONAL,fNodeIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); - } - if (hashIdUpd.isPresent()) { - trans.put(LogicalDatastoreType.OPERATIONAL, groupRef, groupBuilder.build()); + final NodeGroupDescStatsBuilder groupDesc= new NodeGroupDescStatsBuilder(); + groupDesc.setGroupDesc(new GroupDescBuilder(group).build()); + //Update augmented data + groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build()); + existGroupKeys.remove(groupKey); + tx.put(LogicalDatastoreType.OPERATIONAL, groupRef, groupBuilder.build()); } } } - /* Delete all not presented Group Nodes */ - deleteAllNotPresentNode(fNodeIdent, trans, deviceGroupKeys); } private void deleteAllNotPresentNode(final InstanceIdentifier fNodeIdent, final ReadWriteTransaction trans, final List deviceGroupKeys) { - final Optional fNode = readLatestConfiguration(fNodeIdent); - if ( ! fNode.isPresent()) { - LOG.trace("Read Operational/DS for FlowCapableNode fail! Node {} doesn't exist.", fNodeIdent); + Preconditions.checkNotNull(fNodeIdent); + Preconditions.checkNotNull(trans); + + if (deviceGroupKeys == null) { return; } - final List existGroups = fNode.get().getGroup() != null - ? fNode.get().getGroup() : Collections. emptyList(); - /* Add all existed groups paths - no updated paths has to be removed */ - for (final Group group : existGroups) { - if (deviceGroupKeys.remove(group.getKey())) { - break; // group still exist on device - } - LOG.trace("Group {} has to removed.", group); - final InstanceIdentifier delGroupIdent = fNodeIdent.child(Group.class, group.getKey()); + + for (final GroupKey key : deviceGroupKeys) { + final InstanceIdentifier delGroupIdent = fNodeIdent.child(Group.class, key); + LOG.trace("Group {} has to removed.", key); Optional delGroup = Optional.absent(); try { delGroup = trans.read(LogicalDatastoreType.OPERATIONAL, delGroupIdent).checkedGet(); diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitMeter.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitMeter.java index 9c9de59a6a..d6988a6f2b 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitMeter.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitMeter.java @@ -16,6 +16,7 @@ import org.opendaylight.controller.md.sal.binding.api.DataBroker; import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; +import org.opendaylight.controller.md.statistics.manager.StatPermCollector.StatCapabTypes; import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager.TransactionCacheContainer; import org.opendaylight.controller.md.statistics.manager.StatisticsManager; import org.opendaylight.controller.md.statistics.manager.StatisticsManager.StatDataStoreOperation; @@ -36,7 +37,9 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111. import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStats; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStatsBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures; +import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeaturesBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatistics; +import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatisticsBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsListener; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeatures; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeaturesBuilder; @@ -45,11 +48,13 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111. import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatisticsBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats; +import org.opendaylight.yangtools.yang.binding.DataObject; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Optional; +import com.google.common.base.Preconditions; /** * statistics-manager @@ -90,68 +95,102 @@ public class StatListenCommitMeter extends StatAbstractListenCommit meterConfStat = notification.getMeterConfigStats() != null - ? new ArrayList<>(notification.getMeterConfigStats()) : new ArrayList(10); - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if (txContainer.isPresent()) { - final List cacheNotifs = txContainer.get().getNotifications(); - for (final TransactionAware notif : cacheNotifs) { - if (notif instanceof MeterConfigStatsUpdated) { - meterConfStat.addAll(((MeterConfigStatsUpdated) notif).getMeterConfigStats()); - } - } - } - final InstanceIdentifier nodeIdent = InstanceIdentifier.create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { + + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + /* Validate exist FlowCapableNode */ + final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); + Optional fNode = Optional.absent(); + try { + fNode = tx.read(LogicalDatastoreType.OPERATIONAL,fNodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); + } + if ( ! fNode.isPresent()) { + return; + } + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; + } + /* Prepare List actual Meters and not updated Meters will be removed */ + final List existMeters = fNode.get().getMeter() != null + ? fNode.get().getMeter() : Collections. emptyList(); + final List existMeterKeys = new ArrayList<>(); + for (final Meter meter : existMeters) { + existMeterKeys.add(meter.getKey()); + } + /* MeterConfig processing */ + comitConfMeterStats(txContainer, tx, fNodeIdent, existMeterKeys); + /* Delete all not presented Meter Nodes */ + deleteAllNotPresentedNodes(fNodeIdent, tx, Collections.unmodifiableList(existMeterKeys)); /* Notification for continue collecting statistics */ notifyToCollectNextStatistics(nodeIdent); - comitConfMeterStats(meterConfStat, nodeIdent, tx); } }); } @Override public void onMeterFeaturesUpdated(final MeterFeaturesUpdated notification) { + Preconditions.checkNotNull(notification); final TransactionId transId = notification.getTransactionId(); final NodeId nodeId = notification.getId(); if ( ! isExpectedStatistics(transId, nodeId)) { LOG.debug("STAT-MANAGER - MeterFeaturesUpdated: unregistred notification detect TransactionId {}", transId); return; } + manager.getRpcMsgManager().addNotification(notification, nodeId); if (notification.isMoreReplies()) { - manager.getRpcMsgManager().addNotification(notification, nodeId); - return; - } - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if ( ! txContainer.isPresent()) { return; } - final MeterFeatures stats = new MeterFeaturesBuilder(notification).build(); - final InstanceIdentifier nodeIdent = InstanceIdentifier - .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); - final InstanceIdentifier meterFeatureIdent = nodeIdent - .augmentation(NodeMeterFeatures.class).child(MeterFeatures.class); + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { - /* Notification for continue collecting statistics */ - notifyToCollectNextStatistics(nodeIdent); - Optional node = Optional.absent(); - try { - node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; } - if (node.isPresent()) { - tx.put(LogicalDatastoreType.OPERATIONAL, meterFeatureIdent, stats); + + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + final List cacheNotifs = txContainer.get().getNotifications(); + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof MeterFeaturesUpdated)) { + break; + } + final MeterFeatures stats = new MeterFeaturesBuilder((MeterFeaturesUpdated)notif).build(); + final InstanceIdentifier nodeMeterFeatureIdent = + nodeIdent.augmentation(NodeMeterFeatures.class); + final InstanceIdentifier meterFeatureIdent = nodeMeterFeatureIdent + .child(MeterFeatures.class); + Optional node = Optional.absent(); + try { + node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + } + if (node.isPresent()) { + tx.merge(LogicalDatastoreType.OPERATIONAL, nodeMeterFeatureIdent, new NodeMeterFeaturesBuilder().build(), true); + tx.put(LogicalDatastoreType.OPERATIONAL, meterFeatureIdent, stats); + manager.registerAdditionalNodeFeature(nodeIdent, StatCapabTypes.METER_STATS); + } } } }); @@ -159,123 +198,149 @@ public class StatListenCommitMeter extends StatAbstractListenCommit meterStat = notification.getMeterStats() != null - ? new ArrayList<>(notification.getMeterStats()) : new ArrayList(10); - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if (txContainer.isPresent()) { - final List cacheNotifs = txContainer.get().getNotifications(); - for (final TransactionAware notif : cacheNotifs) { - if (notif instanceof MeterConfigStatsUpdated) { - meterStat.addAll(((MeterStatisticsUpdated) notif).getMeterStats()); - } - } - } - final InstanceIdentifier nodeIdent = InstanceIdentifier.create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override public void applyOperation(final ReadWriteTransaction tx) { - statMeterCommit(meterStat, nodeIdent, tx); - /* Notification for continue collecting statistics */ - notifyToCollectNextStatistics(nodeIdent); + + final InstanceIdentifier nodeIdent = InstanceIdentifier + .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); + /* Node exist check */ + Optional node = Optional.absent(); + try { + node = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + } + if ( ! node.isPresent()) { + return; + } + + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; + } + final List cacheNotifs = txContainer.get().getNotifications(); + + Optional notifMeter = Optional.absent(); + final Optional inputObj = txContainer.get().getConfInput(); + if (inputObj.isPresent() && inputObj.get() instanceof Meter) { + notifMeter = Optional. of((Meter)inputObj.get()); + } + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof MeterStatisticsUpdated)) { + break; + } + statMeterCommit(((MeterStatisticsUpdated) notif).getMeterStats(), nodeIdent, tx); + } + if (notifMeter.isPresent()) { + notifyToCollectNextStatistics(nodeIdent); + } } }); } private void statMeterCommit(final List meterStats, - final InstanceIdentifier nodeIdent, final ReadWriteTransaction trans) { + final InstanceIdentifier nodeIdent, final ReadWriteTransaction tx) { + + Preconditions.checkNotNull(meterStats); + Preconditions.checkNotNull(nodeIdent); + Preconditions.checkNotNull(tx); final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); + for (final MeterStats mStat : meterStats) { final MeterStatistics stats = new MeterStatisticsBuilder(mStat).build(); - final MeterKey mKey = new MeterKey(mStat.getMeterId()); - final InstanceIdentifier msIdent = fNodeIdent - .child(Meter.class, mKey).augmentation(NodeMeterStatistics.class) - .child(MeterStatistics.class); + final InstanceIdentifier meterIdent = fNodeIdent.child(Meter.class, new MeterKey(mStat.getMeterId())); + final InstanceIdentifier nodeMeterStatIdent = meterIdent + .augmentation(NodeMeterStatistics.class); + final InstanceIdentifier msIdent = nodeMeterStatIdent.child(MeterStatistics.class); /* Meter Statistics commit */ - Optional fNode = Optional.absent(); + Optional meter = Optional.absent(); try { - fNode = trans.read(LogicalDatastoreType.OPERATIONAL, fNodeIdent).checkedGet(); + meter = tx.read(LogicalDatastoreType.OPERATIONAL, meterIdent).checkedGet(); } catch (final ReadFailedException e) { LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); } - if (fNode.isPresent()) { - trans.put(LogicalDatastoreType.OPERATIONAL, msIdent, stats); + if (meter.isPresent()) { + tx.merge(LogicalDatastoreType.OPERATIONAL, nodeMeterStatIdent, new NodeMeterStatisticsBuilder().build(), true); + tx.put(LogicalDatastoreType.OPERATIONAL, msIdent, stats); } } } - private void comitConfMeterStats(final List meterConfStat, - final InstanceIdentifier nodeIdent, final ReadWriteTransaction trans) { + private void comitConfMeterStats(final Optional> txContainer, final ReadWriteTransaction tx, + final InstanceIdentifier fNodeIdent, final List existMeterKeys) { - final InstanceIdentifier fNodeIdent = nodeIdent.augmentation(FlowCapableNode.class); - final List deviceMeterKeys = new ArrayList<>(); - - for (final MeterConfigStats meterConf : meterConfStat) { - final MeterBuilder meterBuilder = new MeterBuilder(meterConf); - if (meterConf.getMeterId() != null) { - final MeterKey meterKey = new MeterKey(meterConf.getMeterId()); - meterBuilder.setKey(meterKey); - final InstanceIdentifier meterRef = nodeIdent - .augmentation(FlowCapableNode.class).child(Meter.class,meterKey); - final NodeMeterConfigStatsBuilder meterConfig = new NodeMeterConfigStatsBuilder(); - meterConfig.setMeterConfigStats(new MeterConfigStatsBuilder(meterConf).build()); - //Update augmented data - meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build()); - deviceMeterKeys.add(meterKey); - Optional fNode = Optional.absent(); - try { - fNode = trans.read(LogicalDatastoreType.OPERATIONAL, fNodeIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for FlowCapableNode fail! {}", fNodeIdent, e); - } - if (fNode.isPresent()) { - trans.put(LogicalDatastoreType.OPERATIONAL, meterRef, meterBuilder.build()); + Preconditions.checkNotNull(existMeterKeys); + Preconditions.checkNotNull(txContainer); + Preconditions.checkNotNull(fNodeIdent); + Preconditions.checkNotNull(tx); + + final List cacheNotifs = txContainer.get().getNotifications(); + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof MeterConfigStatsUpdated)) { + break; + } + final List meterStats = ((MeterConfigStatsUpdated) notif).getMeterConfigStats(); + if (meterStats == null) { + break; + } + for (final MeterConfigStats meterStat : meterStats) { + if (meterStat.getMeterId() != null) { + final MeterBuilder meterBuilder = new MeterBuilder(meterStat); + final MeterKey meterKey = new MeterKey(meterStat.getMeterId()); + final InstanceIdentifier meterRef = fNodeIdent.child(Meter.class, meterKey); + + final NodeMeterConfigStatsBuilder meterConfig = new NodeMeterConfigStatsBuilder(); + meterConfig.setMeterConfigStats(new MeterConfigStatsBuilder(meterStat).build()); + //Update augmented data + meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build()); + existMeterKeys.remove(meterKey); + tx.put(LogicalDatastoreType.OPERATIONAL, meterRef, meterBuilder.build()); } } } - /* Delete all not presented Meter Nodes */ - deleteAllNotPresentedNodes(fNodeIdent, trans, deviceMeterKeys); } private void deleteAllNotPresentedNodes(final InstanceIdentifier fNodeIdent, - final ReadWriteTransaction trans, final List deviceMeterKeys) { - /* Delete all not presented meters */ - final Optional fNode = readLatestConfiguration(fNodeIdent); + final ReadWriteTransaction tx, final List deviceMeterKeys) { + + Preconditions.checkNotNull(fNodeIdent); + Preconditions.checkNotNull(tx); - if ( ! fNode.isPresent()) { - LOG.trace("Read Operational/DS for FlowCapableNode fail! Node {} doesn't exist.", fNodeIdent); + if (deviceMeterKeys == null) { return; } - final List existMeters = fNode.get().getMeter() != null - ? fNode.get().getMeter() : Collections. emptyList(); - /* Add all existed groups paths - no updated paths has to be removed */ - for (final Meter meter : existMeters) { - if (deviceMeterKeys.remove(meter.getKey())) { - break; // Meter still exist on device - } - final InstanceIdentifier delMeterIdent = fNodeIdent.child(Meter.class, meter.getKey()); + + for (final MeterKey key : deviceMeterKeys) { + final InstanceIdentifier delMeterIdent = fNodeIdent.child(Meter.class, key); + LOG.trace("Meter {} has to removed.", key); Optional delMeter = Optional.absent(); try { - delMeter = trans.read(LogicalDatastoreType.OPERATIONAL, delMeterIdent).checkedGet(); + delMeter = tx.read(LogicalDatastoreType.OPERATIONAL, delMeterIdent).checkedGet(); } catch (final ReadFailedException e) { // NOOP - probably another transaction delete that node } if (delMeter.isPresent()) { - trans.delete(LogicalDatastoreType.OPERATIONAL, delMeterIdent); + tx.delete(LogicalDatastoreType.OPERATIONAL, delMeterIdent); } } } diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitQueue.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitQueue.java index e336f01874..07e167d1e4 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitQueue.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatListenCommitQueue.java @@ -8,8 +8,11 @@ package org.opendaylight.controller.md.statistics.manager.impl; -import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Map.Entry; import org.opendaylight.controller.md.sal.binding.api.DataBroker; import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction; @@ -42,6 +45,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Optional; +import com.google.common.base.Preconditions; /** * statistics-manager @@ -84,65 +88,120 @@ public class StatListenCommitQueue extends StatAbstractListenCommit queueStats = notification.getQueueIdAndStatisticsMap() != null - ? new ArrayList<>(notification.getQueueIdAndStatisticsMap()) : new ArrayList(10); - final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); - if (txContainer.isPresent()) { - final List cachedNotifs = - txContainer.get().getNotifications(); - for (final TransactionAware notif : cachedNotifs) { - if (notif instanceof QueueStatisticsUpdate) { - queueStats.addAll(((QueueStatisticsUpdate) notif).getQueueIdAndStatisticsMap()); - } - } - } - final InstanceIdentifier nodeIdent = InstanceIdentifier.create(Nodes.class) - .child(Node.class, new NodeKey(nodeId)); - /* Queue statistics are small size and we are not able to change for OF cross controller - * - don't need to make are atomic */ + + /* Don't block RPC Notification thread */ manager.enqueue(new StatDataStoreOperation() { @Override - public void applyOperation(final ReadWriteTransaction trans) { - /* Notification for continue */ + public void applyOperation(final ReadWriteTransaction tx) { + + final InstanceIdentifier nodeIdent = InstanceIdentifier.create(Nodes.class) + .child(Node.class, new NodeKey(nodeId)); + + /* Validate exist Node */ + Optional fNode = Optional.absent(); + try { + fNode = tx.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); + } + catch (final ReadFailedException e) { + LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); + } + if ( ! fNode.isPresent()) { + LOG.trace("Read Operational/DS for Node fail! Node {} doesn't exist.", nodeIdent); + return; + } + + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; + } + /* Prepare List actual Queues and not updated Queues will be removed */ + final List existConnectors = fNode.get().getNodeConnector() != null + ? fNode.get().getNodeConnector() : Collections. emptyList(); + final Map existQueueKeys = new HashMap<>(); + for (final NodeConnector connect : existConnectors) { + final List listQueues = connect.getAugmentation(FlowCapableNodeConnector.class).getQueue(); + if (listQueues != null) { + for (final Queue queue : listQueues) { + existQueueKeys.put(queue.getKey(), connect.getKey()); + } + } + } + /* Queue processing */ + statQueueCommit(txContainer, tx, nodeIdent, existQueueKeys); + /* Delete all not presented Group Nodes */ + deleteAllNotPresentedNodes(nodeIdent, tx, Collections.unmodifiableMap(existQueueKeys)); + /* Notification for continue collecting statistics */ notifyToCollectNextStatistics(nodeIdent); - statQueueCommit(queueStats, nodeIdent, trans); } }); } - private void statQueueCommit(final List queueStats, - final InstanceIdentifier nodeIdent, final ReadWriteTransaction trans) { + private void statQueueCommit( + final Optional> txContainer, final ReadWriteTransaction tx, + final InstanceIdentifier nodeIdent, final Map existQueueKeys) { - /* check exist FlowCapableNode and write statistics */ - Optional fNode = Optional.absent(); - try { - fNode = trans.read(LogicalDatastoreType.OPERATIONAL, nodeIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); - return; + Preconditions.checkNotNull(existQueueKeys); + Preconditions.checkNotNull(txContainer); + Preconditions.checkNotNull(nodeIdent); + Preconditions.checkNotNull(tx); + + final List cacheNotifs = txContainer.get().getNotifications(); + for (final TransactionAware notif : cacheNotifs) { + if ( ! (notif instanceof QueueStatisticsUpdate)) { + break; + } + final List queueStats = ((QueueStatisticsUpdate) notif).getQueueIdAndStatisticsMap(); + if (queueStats == null) { + break; + } + for (final QueueIdAndStatisticsMap queueStat : queueStats) { + if (queueStat.getQueueId() != null) { + final FlowCapableNodeConnectorQueueStatistics statChild = + new FlowCapableNodeConnectorQueueStatisticsBuilder(queueStat).build(); + final FlowCapableNodeConnectorQueueStatisticsDataBuilder statBuild = + new FlowCapableNodeConnectorQueueStatisticsDataBuilder(); + statBuild.setFlowCapableNodeConnectorQueueStatistics(statChild); + final QueueKey qKey = new QueueKey(queueStat.getQueueId()); + final InstanceIdentifier queueStatIdent = nodeIdent + .child(NodeConnector.class, new NodeConnectorKey(queueStat.getNodeConnectorId())) + .augmentation(FlowCapableNodeConnector.class) + .child(Queue.class, qKey).augmentation(FlowCapableNodeConnectorQueueStatisticsData.class); + existQueueKeys.remove(qKey); + tx.put(LogicalDatastoreType.OPERATIONAL, queueStatIdent, statBuild.build()); + } + } } - if ( ! fNode.isPresent()) { - LOG.trace("Read Operational/DS for Node fail! Node {} doesn't exist.", nodeIdent); + } + + private void deleteAllNotPresentedNodes(final InstanceIdentifier nodeIdent, + final ReadWriteTransaction tx, final Map existQueueKeys) { + + Preconditions.checkNotNull(nodeIdent); + Preconditions.checkNotNull(tx); + + if (existQueueKeys == null) { return; } - for (final QueueIdAndStatisticsMap queueEntry : queueStats) { - final FlowCapableNodeConnectorQueueStatistics statChild = - new FlowCapableNodeConnectorQueueStatisticsBuilder(queueEntry).build(); - final FlowCapableNodeConnectorQueueStatisticsDataBuilder statBuild = - new FlowCapableNodeConnectorQueueStatisticsDataBuilder(); - statBuild.setFlowCapableNodeConnectorQueueStatistics(statChild); - final QueueKey qKey = new QueueKey(queueEntry.getQueueId()); - final InstanceIdentifier queueStatIdent = nodeIdent - .child(NodeConnector.class, new NodeConnectorKey(queueEntry.getNodeConnectorId())) - .augmentation(FlowCapableNodeConnector.class) - .child(Queue.class, qKey).augmentation(FlowCapableNodeConnectorQueueStatisticsData.class); - trans.put(LogicalDatastoreType.OPERATIONAL, queueStatIdent, statBuild.build()); + for (final Entry entry : existQueueKeys.entrySet()) { + final InstanceIdentifier queueIdent = nodeIdent.child(NodeConnector.class, entry.getValue()) + .augmentation(FlowCapableNodeConnector.class).child(Queue.class, entry.getKey()); + LOG.trace("Queue {} has to removed.", queueIdent); + Optional delQueue = Optional.absent(); + try { + delQueue = tx.read(LogicalDatastoreType.OPERATIONAL, queueIdent).checkedGet(); + } + catch (final ReadFailedException e) { + // NOOP - probably another transaction delete that node + } + if (delQueue.isPresent()) { + tx.delete(LogicalDatastoreType.OPERATIONAL, queueIdent); + } } } } diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatNodeRegistrationImpl.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatNodeRegistrationImpl.java index adb5870986..afb45e59f0 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatNodeRegistrationImpl.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatNodeRegistrationImpl.java @@ -11,12 +11,14 @@ package org.opendaylight.controller.md.statistics.manager.impl; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Set; import org.opendaylight.controller.md.sal.binding.api.DataBroker; import org.opendaylight.controller.md.sal.binding.api.DataChangeListener; import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction; +import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; +import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; -import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; import org.opendaylight.controller.md.statistics.manager.StatNodeRegistration; import org.opendaylight.controller.md.statistics.manager.StatPermCollector.StatCapabTypes; import org.opendaylight.controller.md.statistics.manager.StatisticsManager; @@ -36,9 +38,10 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeCon import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemoved; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; -import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures; import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.yang.binding.DataObject; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,7 +63,7 @@ import com.google.common.base.Preconditions; * * Created: Aug 28, 2014 */ -public class StatNodeRegistrationImpl implements StatNodeRegistration { +public class StatNodeRegistrationImpl implements StatNodeRegistration, DataChangeListener { private static final Logger LOG = LoggerFactory.getLogger(StatNodeRegistrationImpl.class); @@ -74,6 +77,11 @@ public class StatNodeRegistrationImpl implements StatNodeRegistration { Preconditions.checkArgument(db != null, "DataBroker can not be null!"); Preconditions.checkArgument(notificationService != null, "NotificationProviderService can not be null!"); notifListenerRegistration = notificationService.registerNotificationListener(this); + /* Build Path */ + final InstanceIdentifier flowNodeWildCardIdentifier = InstanceIdentifier.create(Nodes.class) + .child(Node.class).augmentation(FlowCapableNode.class); + listenerRegistration = db.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, + flowNodeWildCardIdentifier, StatNodeRegistrationImpl.this, DataChangeScope.BASE); } @Override @@ -131,21 +139,6 @@ public class StatNodeRegistrationImpl implements StatNodeRegistration { maxCapTables = data.getMaxTables(); final Optional maxTables = Optional. of(maxCapTables); - - /* Meters management */ - final InstanceIdentifier meterFeaturesIdent = nodeIdent.augmentation(NodeMeterFeatures.class); - - - Optional meterFeatures = Optional.absent(); - try { - meterFeatures = tx.read(LogicalDatastoreType.OPERATIONAL, meterFeaturesIdent).checkedGet(); - } - catch (final ReadFailedException e) { - LOG.warn("Read NodeMeterFeatures {} fail!", meterFeaturesIdent, e); - } - if (meterFeatures.isPresent()) { - statCapabTypes.add(StatCapabTypes.METER_STATS); - } manager.connectedNodeRegistration(nodeIdent, Collections.unmodifiableList(statCapabTypes), maxTables.get()); } @@ -178,6 +171,7 @@ public class StatNodeRegistrationImpl implements StatNodeRegistration { @Override public void onNodeRemoved(final NodeRemoved notification) { + Preconditions.checkNotNull(notification); final NodeRef nodeRef = notification.getNodeRef(); final InstanceIdentifier nodeRefIdent = nodeRef.getValue(); final InstanceIdentifier nodeIdent = @@ -189,6 +183,7 @@ public class StatNodeRegistrationImpl implements StatNodeRegistration { @Override public void onNodeUpdated(final NodeUpdated notification) { + Preconditions.checkNotNull(notification); final FlowCapableNodeUpdated newFlowNode = notification.getAugmentation(FlowCapableNodeUpdated.class); if (newFlowNode != null && newFlowNode.getSwitchFeatures() != null) { @@ -203,5 +198,25 @@ public class StatNodeRegistrationImpl implements StatNodeRegistration { connectFlowCapableNode(swichFeaturesIdent, switchFeatures, nodeIdent); } } + + @Override + public void onDataChanged(final AsyncDataChangeEvent, DataObject> changeEvent) { + Preconditions.checkNotNull(changeEvent,"Async ChangeEvent can not be null!"); + /* All DataObjects for create */ + final Set> createdData = changeEvent.getCreatedData() != null + ? changeEvent.getCreatedData().keySet() : Collections.> emptySet(); + + for (final InstanceIdentifier entryKey : createdData) { + final InstanceIdentifier nodeIdent = entryKey + .firstIdentifierOf(Node.class); + if ( ! nodeIdent.isWildcarded()) { + final NodeRef nodeRef = new NodeRef(nodeIdent); + // FIXME: these calls is a job for handshake or for inventory manager + /* check Group and Meter future */ + manager.getRpcMsgManager().getGroupFeaturesStat(nodeRef); + manager.getRpcMsgManager().getMeterFeaturesStat(nodeRef); + } + } + } } diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatPermCollectorImpl.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatPermCollectorImpl.java index d008042e8d..ff1778e8aa 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatPermCollectorImpl.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatPermCollectorImpl.java @@ -1,5 +1,6 @@ package org.opendaylight.controller.md.statistics.manager.impl; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -80,6 +81,11 @@ public class StatPermCollectorImpl implements StatPermCollector { statNetCollectorServ.shutdown(); } + @Override + public boolean hasActiveNodes() { + return ( ! statNodeHolder.isEmpty()); + } + @Override public boolean isProvidedFlowNodeActive( final InstanceIdentifier flowNode) { @@ -89,9 +95,7 @@ public class StatPermCollectorImpl implements StatPermCollector { @Override public boolean connectedNodeRegistration(final InstanceIdentifier ident, final List statTypes, final Short nrOfSwitchTables) { - if (ident.isWildcarded()) { - LOG.warn("FlowCapableNode IstanceIdentifier {} registration can not be wildcarded!", ident); - } else { + if (isNodeIdentValidForUse(ident)) { if ( ! statNodeHolder.containsKey(ident)) { synchronized (statNodeHolderLock) { final boolean startStatCollecting = statNodeHolder.size() == 0; @@ -119,9 +123,7 @@ public class StatPermCollectorImpl implements StatPermCollector { @Override public boolean disconnectedNodeUnregistration(final InstanceIdentifier ident) { - if (ident.isWildcarded()) { - LOG.warn("FlowCapableNode IstanceIdentifier {} unregistration can not be wildcarded!", ident); - } else { + if (isNodeIdentValidForUse(ident)) { if (statNodeHolder.containsKey(ident)) { synchronized (statNodeHolderLock) { if (statNodeHolder.containsKey(ident)) { @@ -142,6 +144,33 @@ public class StatPermCollectorImpl implements StatPermCollector { return false; } + @Override + public boolean registerAdditionalNodeFeature(final InstanceIdentifier ident, + final StatCapabTypes statCapab) { + if (isNodeIdentValidForUse(ident)) { + if ( ! statNodeHolder.containsKey(ident)) { + return false; + } + final StatNodeInfoHolder statNode = statNodeHolder.get(ident); + if ( ! statNode.getStatMarkers().contains(statCapab)) { + synchronized (statNodeHolderLock) { + if ( ! statNode.getStatMarkers().contains(statCapab)) { + final List statCapabForEdit = new ArrayList<>(statNode.getStatMarkers()); + statCapabForEdit.add(statCapab); + final StatNodeInfoHolder nodeInfoHolder = new StatNodeInfoHolder(statNode.getNodeRef(), + Collections.unmodifiableList(statCapabForEdit), statNode.getMaxTables()); + + final Map, StatNodeInfoHolder> statNodes = + new HashMap<>(statNodeHolder); + statNodes.put(ident, nodeInfoHolder); + statNodeHolder = Collections.unmodifiableMap(statNodes); + } + } + } + } + return true; + } + @Override public void collectNextStatistics() { if (wakeMe) { @@ -238,8 +267,6 @@ public class StatPermCollectorImpl implements StatPermCollector { break; case GROUP_STATS: LOG.trace("STAT-MANAGER-collecting GROUP-STATS for NodeRef {}", actualNodeRef); - manager.getRpcMsgManager().getGroupFeaturesStat(actualNodeRef); - waitingForNotification(); manager.getRpcMsgManager().getAllGroupsConfStats(actualNodeRef); waitingForNotification(); manager.getRpcMsgManager().getAllGroupsStat(actualNodeRef); @@ -247,8 +274,6 @@ public class StatPermCollectorImpl implements StatPermCollector { break; case METER_STATS: LOG.trace("STAT-MANAGER-collecting METER-STATS for NodeRef {}", actualNodeRef); - manager.getRpcMsgManager().getMeterFeaturesStat(actualNodeRef); - waitingForNotification(); manager.getRpcMsgManager().getAllMeterConfigStat(actualNodeRef); waitingForNotification(); manager.getRpcMsgManager().getAllMetersStat(actualNodeRef); @@ -297,9 +322,16 @@ public class StatPermCollectorImpl implements StatPermCollector { } } - @Override - public boolean hasActiveNodes() { - return ( ! statNodeHolder.isEmpty()); + private boolean isNodeIdentValidForUse(final InstanceIdentifier ident) { + if (ident == null) { + LOG.warn("FlowCapableNode InstanceIdentifier {} can not be null!"); + return false; + } + if (ident.isWildcarded()) { + LOG.warn("FlowCapableNode InstanceIdentifier {} can not be wildcarded!", ident); + return false; + } + return true; } } diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatRpcMsgManagerImpl.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatRpcMsgManagerImpl.java index e53f494129..176e52708b 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatRpcMsgManagerImpl.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatRpcMsgManagerImpl.java @@ -454,7 +454,6 @@ public class StatRpcMsgManagerImpl implements StatRpcMsgManager { @Override public Void call() throws Exception { - Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!"); final GetGroupDescriptionInputBuilder builder = new GetGroupDescriptionInputBuilder(); builder.setNode(nodeRef); diff --git a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatisticsManagerImpl.java b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatisticsManagerImpl.java index ca0e5b20ed..396ff0e1f6 100644 --- a/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatisticsManagerImpl.java +++ b/opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatisticsManagerImpl.java @@ -300,7 +300,18 @@ public class StatisticsManagerImpl implements StatisticsManager, Runnable { return; } } - LOG.debug("Node {} has not removed.", nodeIdent); + LOG.debug("Node {} has not been removed.", nodeIdent); + } + + @Override + public void registerAdditionalNodeFeature(final InstanceIdentifier nodeIdent, + final StatCapabTypes statCapab) { + for (final StatPermCollector collector : statCollectors) { + if (collector.registerAdditionalNodeFeature(nodeIdent, statCapab)) { + return; + } + } + LOG.debug("Node {} has not been extended for feature {}!", nodeIdent, statCapab); } /* Getter internal Statistic Manager Job Classes */ diff --git a/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/Main.java b/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/Main.java index 79c083c1ed..64397de118 100644 --- a/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/Main.java +++ b/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/Main.java @@ -7,6 +7,8 @@ */ package org.opendaylight.controller.netconf.cli; +import static com.google.common.base.Throwables.getStackTraceAsString; + import com.google.common.base.Preconditions; import java.io.IOException; import java.net.InetAddress; @@ -105,7 +107,7 @@ public class Main { } private static void handleException(final Exception e, final String message) { - // FIXME syserr the exception and stacktrace + System.console().writer().println(String.format("Error %s cause %s", message, getStackTraceAsString(e.fillInStackTrace()))); } private static void writeStatus(final ConsoleIO io, final String blueprint, final Object... args) { diff --git a/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/NetconfDeviceConnectionManager.java b/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/NetconfDeviceConnectionManager.java index 3dd892e169..f4e21714cd 100644 --- a/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/NetconfDeviceConnectionManager.java +++ b/opendaylight/netconf/netconf-cli/src/main/java/org/opendaylight/controller/netconf/cli/NetconfDeviceConnectionManager.java @@ -22,12 +22,21 @@ import org.opendaylight.controller.netconf.cli.io.ConsoleIO; import org.opendaylight.controller.netconf.client.NetconfClientDispatcherImpl; import org.opendaylight.controller.netconf.client.conf.NetconfClientConfigurationBuilder; import org.opendaylight.controller.sal.connect.netconf.NetconfDevice; +import org.opendaylight.controller.sal.connect.netconf.NetconfDevice.SchemaResourcesDTO; +import org.opendaylight.controller.sal.connect.netconf.NetconfStateSchemas.NetconfStateSchemasResolverImpl; import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator; +import org.opendaylight.controller.sal.connect.netconf.schema.mapping.NetconfMessageTransformer; import org.opendaylight.controller.sal.connect.util.RemoteDeviceId; +import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory; +import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter; +import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource; +import org.opendaylight.yangtools.yang.model.repo.util.FilesystemSchemaSourceCache; import org.opendaylight.yangtools.yang.model.util.repo.AbstractCachingSchemaSourceProvider; import org.opendaylight.yangtools.yang.model.util.repo.FilesystemSchemaCachingProvider; import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProvider; import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProviders; +import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository; +import org.opendaylight.yangtools.yang.parser.util.TextToASTTransformer; /** * Manages connect/disconnect to 1 remote device @@ -42,6 +51,8 @@ public class NetconfDeviceConnectionManager implements Closeable { private final NioEventLoopGroup nettyThreadGroup; private final NetconfClientDispatcherImpl netconfClientDispatcher; + private static final String CACHE = "cache/schema"; + // Connection private NetconfDeviceConnectionHandler handler; private NetconfDevice device; @@ -70,7 +81,15 @@ public class NetconfDeviceConnectionManager implements Closeable { handler = new NetconfDeviceConnectionHandler(commandDispatcher, schemaContextRegistry, console, name); - device = NetconfDevice.createNetconfDevice(deviceId, getGlobalNetconfSchemaProvider(), executor, handler); + + final SharedSchemaRepository repository = new SharedSchemaRepository("repo"); + final SchemaContextFactory schemaContextFactory = repository.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT); + final FilesystemSchemaSourceCache cache = new FilesystemSchemaSourceCache<>(repository, YangTextSchemaSource.class, new File(CACHE)); + repository.registerSchemaSourceListener(cache); + repository.registerSchemaSourceListener(TextToASTTransformer.create(repository, repository)); + + device = new NetconfDevice(new SchemaResourcesDTO(repository, schemaContextFactory, new NetconfStateSchemasResolverImpl()), + deviceId, handler, executor, new NetconfMessageTransformer()); listener = new NetconfDeviceCommunicator(deviceId, device); configBuilder.withSessionListener(listener); listener.initializeRemoteConnection(netconfClientDispatcher, configBuilder.build()); diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java index 14d753f1f8..c8c9128282 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java @@ -137,7 +137,6 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel()); connectPromise.setSuccess(); - connectPromise = null; // TODO we should also read from error stream and at least log from that @@ -162,9 +161,12 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable e) { LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), e); - connectPromise.setFailure(e); - connectPromise = null; - throw new IllegalStateException("Unable to setup SSH connection on channel: " + ctx.channel(), e); + disconnect(ctx, ctx.newPromise()); + + // If the promise is not yet done, we have failed with initial connect and set connectPromise to failure + if(!connectPromise.isDone()) { + connectPromise.setFailure(e); + } } @Override @@ -185,6 +187,15 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { @Override public synchronized void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) { + // Super disconnect is necessary in this case since we are using NioSocketChannel and it needs to cleanup its resources + // e.g. Socket that it tries to open in its constructor (https://bugs.opendaylight.org/show_bug.cgi?id=2430) + // TODO better solution would be to implement custom ChannelFactory + Channel that will use mina SSH lib internally: port this to custom channel implementation + try { + super.disconnect(ctx, ctx.newPromise()); + } catch (final Exception e) { + LOG.warn("Unable to cleanup all resources for channel: {}. Ignoring.", ctx.channel(), e); + } + if(sshReadAsyncListener != null) { sshReadAsyncListener.close(); } @@ -205,11 +216,15 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { }); } + // If we have already succeeded and the session was dropped after, we need to fire inactive to notify reconnect logic + if(connectPromise.isSuccess()) { + ctx.fireChannelInactive(); + } + channel = null; - promise.setSuccess(); + promise.setSuccess(); LOG.debug("SSH session closed on channel: {}", ctx.channel()); - ctx.fireChannelInactive(); } } diff --git a/opendaylight/netconf/netconf-netty-util/src/test/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerTest.java b/opendaylight/netconf/netconf-netty-util/src/test/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerTest.java index 73f2287c8b..dfca1b80c4 100644 --- a/opendaylight/netconf/netconf-netty-util/src/test/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerTest.java +++ b/opendaylight/netconf/netconf-netty-util/src/test/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerTest.java @@ -8,7 +8,6 @@ package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; -import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyBoolean; import static org.mockito.Matchers.anyObject; @@ -18,9 +17,9 @@ import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.verifyZeroInteractions; import com.google.common.util.concurrent.FutureCallback; @@ -30,8 +29,10 @@ import com.google.common.util.concurrent.SettableFuture; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; +import io.netty.channel.DefaultChannelPromise; import java.io.IOException; import java.net.SocketAddress; import org.apache.sshd.ClientChannel; @@ -143,6 +144,7 @@ public class AsyncSshHandlerTest { doReturn(ctx).when(ctx).fireChannelActive(); doReturn(ctx).when(ctx).fireChannelInactive(); doReturn(ctx).when(ctx).fireChannelRead(anyObject()); + doReturn(mock(ChannelFuture.class)).when(ctx).disconnect(any(ChannelPromise.class)); doReturn(getMockedPromise()).when(ctx).newPromise(); } @@ -179,7 +181,6 @@ public class AsyncSshHandlerTest { verify(subsystemChannel).setStreaming(ClientChannel.Streaming.Async); verify(promise).setSuccess(); - verifyNoMoreInteractions(promise); verify(ctx).fireChannelActive(); } @@ -533,14 +534,8 @@ public class AsyncSshHandlerTest { verify(subsystemChannel).setStreaming(ClientChannel.Streaming.Async); - try { - sshChannelOpenListener.operationComplete(getFailedOpenFuture()); - fail("Exception expected"); - } catch (final Exception e) { - verify(promise).setFailure(any(Throwable.class)); - verifyNoMoreInteractions(promise); - // TODO should ctx.channelInactive be called if we throw exception ? - } + sshChannelOpenListener.operationComplete(getFailedOpenFuture()); + verify(promise).setFailure(any(Throwable.class)); } @Test @@ -555,14 +550,8 @@ public class AsyncSshHandlerTest { final AuthFuture authFuture = getFailedAuthFuture(); - try { - sshAuthListener.operationComplete(authFuture); - fail("Exception expected"); - } catch (final Exception e) { - verify(promise).setFailure(any(Throwable.class)); - verifyNoMoreInteractions(promise); - // TODO should ctx.channelInactive be called ? - } + sshAuthListener.operationComplete(authFuture); + verify(promise).setFailure(any(Throwable.class)); } private AuthFuture getFailedAuthFuture() { @@ -584,14 +573,8 @@ public class AsyncSshHandlerTest { asyncSshHandler.connect(ctx, remoteAddress, localAddress, promise); final ConnectFuture connectFuture = getFailedConnectFuture(); - try { - sshConnectListener.operationComplete(connectFuture); - fail("Exception expected"); - } catch (final Exception e) { - verify(promise).setFailure(any(Throwable.class)); - verifyNoMoreInteractions(promise); - // TODO should ctx.channelInactive be called ? - } + sshConnectListener.operationComplete(connectFuture); + verify(promise).setFailure(any(Throwable.class)); } private ConnectFuture getFailedConnectFuture() { @@ -602,10 +585,7 @@ public class AsyncSshHandlerTest { } private ChannelPromise getMockedPromise() { - final ChannelPromise promise = mock(ChannelPromise.class); - doReturn(promise).when(promise).setSuccess(); - doReturn(promise).when(promise).setFailure(any(Throwable.class)); - return promise; + return spy(new DefaultChannelPromise(channel)); } private static abstract class SuccessFutureListener> implements FutureCallback> { diff --git a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java index 2802488170..72534e242e 100644 --- a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java +++ b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java @@ -136,7 +136,7 @@ public class SSHTest { Thread.sleep(100); } assertFalse(echoClientHandler.isConnected()); - assertEquals(State.CONNECTION_CLOSED, echoClientHandler.getState()); + assertEquals(State.FAILED_TO_CONNECT, echoClientHandler.getState()); } } diff --git a/opendaylight/netconf/pom.xml b/opendaylight/netconf/pom.xml index 885bdb4fdd..8b14adf252 100644 --- a/opendaylight/netconf/pom.xml +++ b/opendaylight/netconf/pom.xml @@ -19,8 +19,7 @@ netconf-api - - + netconf-cli netconf-config netconf-impl config-netconf-connector diff --git a/opendaylight/networkconfiguration/neutron/northbound/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronLoadBalancerPoolNorthbound.java b/opendaylight/networkconfiguration/neutron/northbound/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronLoadBalancerPoolNorthbound.java index 5235030b4f..f0ae6094d4 100644 --- a/opendaylight/networkconfiguration/neutron/northbound/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronLoadBalancerPoolNorthbound.java +++ b/opendaylight/networkconfiguration/neutron/northbound/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronLoadBalancerPoolNorthbound.java @@ -200,8 +200,6 @@ public class NeutronLoadBalancerPoolNorthbound { if (loadBalancerPoolInterface.neutronLoadBalancerPoolExists(singleton.getLoadBalancerPoolID())) { throw new BadRequestException("LoadBalancerPool UUID already exists"); } - loadBalancerPoolInterface.addNeutronLoadBalancerPool(singleton); - Object[] instances = ServiceHelper.getGlobalInstances(INeutronLoadBalancerPoolAware.class, this, null); if (instances != null) { for (Object instance : instances) {