From: Vaclav Demcak Date: Mon, 27 Oct 2014 23:17:39 +0000 (+0100) Subject: Fix bug 2413 NPE for group and meters X-Git-Tag: release/lithium~830 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=commitdiff_plain;h=227a056ef3f8d9dee9414948ea261b0d2348a1f1 Fix bug 2413 NPE for group and meters * group/meter features * Device RPC features Calls for group/meter depend on FlowCapableNode data change event (create event only) * Remove Meter's and Group's feature calls from perm. stat. collect * Add API methods for additional registration new StatCapabTypes for a specific Node statistic collecting process * Fix NPE for features * refactoring group/meter stats update processing (Target: Don't block the RPC Notification threads as soon as is possible) * Fix NPE List processing * queue - add delete processing (becasue queues has to be edit outside and we are not cleaning it) * add validations input values * remove unnecessary validation input values (StatRpcMsgManagerImpl) Signed-off-by: Vaclav Demcak Change-Id: If10a0ea0bb098ca70cd19ce32a7ca5790159aae5 --- 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 */