X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fstatistics-manager%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fstatistics%2Fmanager%2Fimpl%2FStatListenCommitGroup.java;h=538b9ef50565b8330e0960c0b192019fe9db959d;hp=f351132f7f816bcc98151e9e6cb991c7ea911310;hb=a5899521495f3a3d089f42ed026bd5a6fea62411;hpb=c2f8c5b00708ab1d9e003ede0c960133989d5e0c 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..538b9ef505 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,9 +16,11 @@ 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; +import org.opendaylight.controller.md.statistics.manager.StatisticsManager.StatDataStoreOperation.StatsManagerOperationType; import org.opendaylight.controller.sal.binding.api.NotificationProviderService; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionAware; @@ -29,7 +31,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 +55,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Optional; +import com.google.common.base.Preconditions; /** * statistics-manager @@ -90,72 +95,103 @@ 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)); - manager.enqueue(new StatDataStoreOperation() { + + /* Don't block RPC Notification thread */ + manager.enqueue(new StatDataStoreOperation(StatsManagerOperationType.DATA_COMMIT_OPER_DS,nodeId) { @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); + notifyToCollectNextStatistics(nodeIdent, transId); } }); } @Override public void onGroupFeaturesUpdated(final GroupFeaturesUpdated 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); + LOG.debug("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 InstanceIdentifier nodeIdent = InstanceIdentifier - .create(Nodes.class).child(Node.class, new NodeKey(nodeId)); - manager.enqueue(new StatDataStoreOperation() { + /* Don't block RPC Notification thread */ + manager.enqueue(new StatDataStoreOperation(StatsManagerOperationType.DATA_COMMIT_OPER_DS,nodeId) { @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(); + /* Get and Validate TransactionCacheContainer */ + final Optional> txContainer = getTransactionCacheContainer(transId, nodeId); + if ( ! isTransactionCacheContainerValid(txContainer)) { + return; } - catch (final ReadFailedException e) { - LOG.debug("Read Operational/DS for Node fail! {}", nodeIdent, e); - } - 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 +199,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)); - manager.enqueue(new StatDataStoreOperation() { + + /* Don't block RPC Notification thread */ + manager.enqueue(new StatDataStoreOperation(StatsManagerOperationType.DATA_COMMIT_OPER_DS,nodeId) { @Override public void applyOperation(final ReadWriteTransaction tx) { - /* Notification for continue collecting statistics */ - if ( ! group.isPresent()) { - 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 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, transId); } - 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();