A transaction is additive and there's little harm in commiting empty
ones. Move allocation/commit outside of tight loops, such that we get
proper bulking.
Change-Id: I1f4920965c8ecf3091b6afb69ce1e1baa7ca0e24
Signed-off-by: Robert Varga <rovarga@cisco.com>
import org.slf4j.LoggerFactory;
/**
import org.slf4j.LoggerFactory;
/**
- * Class implement statistics manager related listener interface and augment all the
+ * Class implement statistics manager related listener interface and augment all the
* received statistics data to data stores.
* received statistics data to data stores.
- * TODO: Need to add error message listener and clean-up the associated tx id
+ * TODO: Need to add error message listener and clean-up the associated tx id
* if it exists in the tx-id cache.
* @author vishnoianil
*
*/
public class StatisticsUpdateCommiter implements OpendaylightGroupStatisticsListener,
* if it exists in the tx-id cache.
* @author vishnoianil
*
*/
public class StatisticsUpdateCommiter implements OpendaylightGroupStatisticsListener,
- OpendaylightMeterStatisticsListener,
+ OpendaylightMeterStatisticsListener,
OpendaylightFlowStatisticsListener,
OpendaylightPortStatisticsListener,
OpendaylightFlowTableStatisticsListener,
OpendaylightQueueStatisticsListener{
OpendaylightFlowStatisticsListener,
OpendaylightPortStatisticsListener,
OpendaylightFlowTableStatisticsListener,
OpendaylightQueueStatisticsListener{
private final static Logger sucLogger = LoggerFactory.getLogger(StatisticsUpdateCommiter.class);
private final StatisticsProvider statisticsManager;
private final MultipartMessageManager messageManager;
private final static Logger sucLogger = LoggerFactory.getLogger(StatisticsUpdateCommiter.class);
private final StatisticsProvider statisticsManager;
private final MultipartMessageManager messageManager;
private int unaccountedFlowsCounter = 1;
/**
private int unaccountedFlowsCounter = 1;
/**
this.statisticsManager = manager;
this.messageManager = this.statisticsManager.getMultipartMessageManager();
}
this.statisticsManager = manager;
this.messageManager = this.statisticsManager.getMultipartMessageManager();
}
public StatisticsProvider getStatisticsManager(){
return statisticsManager;
}
public StatisticsProvider getStatisticsManager(){
return statisticsManager;
}
@Override
public void onMeterConfigStatsUpdated(MeterConfigStatsUpdated notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
@Override
public void onMeterConfigStatsUpdated(MeterConfigStatsUpdated notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
//Add statistics to local cache
NodeKey key = new NodeKey(notification.getId());
//Add statistics to local cache
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
cache.get(notification.getId()).updateMeterConfigStats(notification.getMeterConfigStats());
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
cache.get(notification.getId()).updateMeterConfigStats(notification.getMeterConfigStats());
//Publish data to configuration data store
List<MeterConfigStats> meterConfigStatsList = notification.getMeterConfigStats();
//Publish data to configuration data store
List<MeterConfigStats> meterConfigStatsList = notification.getMeterConfigStats();
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for(MeterConfigStats meterConfigStats : meterConfigStatsList){
for(MeterConfigStats meterConfigStats : meterConfigStatsList){
- DataModificationTransaction it = this.statisticsManager.startChange();
MeterBuilder meterBuilder = new MeterBuilder();
MeterKey meterKey = new MeterKey(meterConfigStats.getMeterId());
meterBuilder.setKey(meterKey);
MeterBuilder meterBuilder = new MeterBuilder();
MeterKey meterKey = new MeterKey(meterConfigStats.getMeterId());
meterBuilder.setKey(meterKey);
InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Meter.class,meterKey).toInstance();
InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Meter.class,meterKey).toInstance();
NodeMeterConfigStatsBuilder meterConfig= new NodeMeterConfigStatsBuilder();
MeterConfigStatsBuilder stats = new MeterConfigStatsBuilder();
stats.fieldsFrom(meterConfigStats);
meterConfig.setMeterConfigStats(stats.build());
NodeMeterConfigStatsBuilder meterConfig= new NodeMeterConfigStatsBuilder();
MeterConfigStatsBuilder stats = new MeterConfigStatsBuilder();
stats.fieldsFrom(meterConfigStats);
meterConfig.setMeterConfigStats(stats.build());
//Update augmented data
meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build());
it.putOperationalData(meterRef, meterBuilder.build());
//Update augmented data
meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build());
it.putOperationalData(meterRef, meterBuilder.build());
}
@Override
public void onMeterStatisticsUpdated(MeterStatisticsUpdated notification) {
}
@Override
public void onMeterStatisticsUpdated(MeterStatisticsUpdated notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
+
+ //Publish data to configuration data store
List<MeterStats> meterStatsList = notification.getMeterStats();
List<MeterStats> meterStatsList = notification.getMeterStats();
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for(MeterStats meterStats : meterStatsList){
for(MeterStats meterStats : meterStatsList){
- //Publish data to configuration data store
- DataModificationTransaction it = this.statisticsManager.startChange();
MeterBuilder meterBuilder = new MeterBuilder();
MeterKey meterKey = new MeterKey(meterStats.getMeterId());
meterBuilder.setKey(meterKey);
MeterBuilder meterBuilder = new MeterBuilder();
MeterKey meterKey = new MeterKey(meterStats.getMeterId());
meterBuilder.setKey(meterKey);
InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Meter.class,meterKey).toInstance();
InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Meter.class,meterKey).toInstance();
NodeMeterStatisticsBuilder meterStatsBuilder= new NodeMeterStatisticsBuilder();
MeterStatisticsBuilder stats = new MeterStatisticsBuilder();
stats.fieldsFrom(meterStats);
NodeMeterStatisticsBuilder meterStatsBuilder= new NodeMeterStatisticsBuilder();
MeterStatisticsBuilder stats = new MeterStatisticsBuilder();
stats.fieldsFrom(meterStats);
//Update augmented data
meterBuilder.addAugmentation(NodeMeterStatistics.class, meterStatsBuilder.build());
it.putOperationalData(meterRef, meterBuilder.build());
//Update augmented data
meterBuilder.addAugmentation(NodeMeterStatistics.class, meterStatsBuilder.build());
it.putOperationalData(meterRef, meterBuilder.build());
}
@Override
public void onGroupDescStatsUpdated(GroupDescStatsUpdated notification) {
}
@Override
public void onGroupDescStatsUpdated(GroupDescStatsUpdated notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
cache.get(notification.getId()).updateGroupDescStats(notification.getGroupDescStats());
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
cache.get(notification.getId()).updateGroupDescStats(notification.getGroupDescStats());
//Publish data to configuration data store
List<GroupDescStats> groupDescStatsList = notification.getGroupDescStats();
//Publish data to configuration data store
List<GroupDescStats> groupDescStatsList = notification.getGroupDescStats();
+ DataModificationTransaction it = this.statisticsManager.startChange();
for(GroupDescStats groupDescStats : groupDescStatsList){
for(GroupDescStats groupDescStats : groupDescStatsList){
- DataModificationTransaction it = this.statisticsManager.startChange();
-
GroupBuilder groupBuilder = new GroupBuilder();
GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
groupBuilder.setKey(groupKey);
GroupBuilder groupBuilder = new GroupBuilder();
GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
groupBuilder.setKey(groupKey);
InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Group.class,groupKey).toInstance();
InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Group.class,groupKey).toInstance();
GroupDescBuilder stats = new GroupDescBuilder();
stats.fieldsFrom(groupDescStats);
groupDesc.setGroupDesc(stats.build());
GroupDescBuilder stats = new GroupDescBuilder();
stats.fieldsFrom(groupDescStats);
groupDesc.setGroupDesc(stats.build());
//Update augmented data
groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
it.putOperationalData(groupRef, groupBuilder.build());
//Update augmented data
groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
it.putOperationalData(groupRef, groupBuilder.build());
}
@Override
public void onGroupStatisticsUpdated(GroupStatisticsUpdated notification) {
}
@Override
public void onGroupStatisticsUpdated(GroupStatisticsUpdated notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
//Publish data to configuration data store
NodeKey key = new NodeKey(notification.getId());
List<GroupStats> groupStatsList = notification.getGroupStats();
//Publish data to configuration data store
NodeKey key = new NodeKey(notification.getId());
List<GroupStats> groupStatsList = notification.getGroupStats();
+ DataModificationTransaction it = this.statisticsManager.startChange();
for(GroupStats groupStats : groupStatsList){
for(GroupStats groupStats : groupStatsList){
- DataModificationTransaction it = this.statisticsManager.startChange();
-
GroupBuilder groupBuilder = new GroupBuilder();
GroupKey groupKey = new GroupKey(groupStats.getGroupId());
groupBuilder.setKey(groupKey);
GroupBuilder groupBuilder = new GroupBuilder();
GroupKey groupKey = new GroupKey(groupStats.getGroupId());
groupBuilder.setKey(groupKey);
InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Group.class,groupKey).toInstance();
InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
.augmentation(FlowCapableNode.class)
.child(Group.class,groupKey).toInstance();
GroupStatisticsBuilder stats = new GroupStatisticsBuilder();
stats.fieldsFrom(groupStats);
groupStatisticsBuilder.setGroupStatistics(stats.build());
GroupStatisticsBuilder stats = new GroupStatisticsBuilder();
stats.fieldsFrom(groupStats);
groupStatisticsBuilder.setGroupStatistics(stats.build());
//Update augmented data
groupBuilder.addAugmentation(NodeGroupStatistics.class, groupStatisticsBuilder.build());
it.putOperationalData(groupRef, groupBuilder.build());
//Update augmented data
groupBuilder.addAugmentation(NodeGroupStatistics.class, groupStatisticsBuilder.build());
it.putOperationalData(groupRef, groupBuilder.build());
@Override
public void onMeterFeaturesUpdated(MeterFeaturesUpdated notification) {
@Override
public void onMeterFeaturesUpdated(MeterFeaturesUpdated notification) {
meterFeature.setMaxBands(notification.getMaxBands());
meterFeature.setMaxColor(notification.getMaxColor());
meterFeature.setMaxMeter(notification.getMaxMeter());
meterFeature.setMaxBands(notification.getMaxBands());
meterFeature.setMaxColor(notification.getMaxColor());
meterFeature.setMaxMeter(notification.getMaxMeter());
//Publish data to configuration data store
DataModificationTransaction it = this.statisticsManager.startChange();
NodeKey key = new NodeKey(notification.getId());
NodeRef ref = getNodeRef(key);
//Publish data to configuration data store
DataModificationTransaction it = this.statisticsManager.startChange();
NodeKey key = new NodeKey(notification.getId());
NodeRef ref = getNodeRef(key);
-
- final NodeBuilder nodeData = new NodeBuilder();
+
+ final NodeBuilder nodeData = new NodeBuilder();
NodeMeterFeaturesBuilder nodeMeterFeatures= new NodeMeterFeaturesBuilder();
nodeMeterFeatures.setMeterFeatures(meterFeature.build());
NodeMeterFeaturesBuilder nodeMeterFeatures= new NodeMeterFeaturesBuilder();
nodeMeterFeatures.setMeterFeatures(meterFeature.build());
//Update augmented data
nodeData.addAugmentation(NodeMeterFeatures.class, nodeMeterFeatures.build());
//Update augmented data
nodeData.addAugmentation(NodeMeterFeatures.class, nodeMeterFeatures.build());
InstanceIdentifier<? extends Object> refValue = ref.getValue();
it.putOperationalData(refValue, nodeData.build());
it.commit();
}
InstanceIdentifier<? extends Object> refValue = ref.getValue();
it.putOperationalData(refValue, nodeData.build());
it.commit();
}
@Override
public void onGroupFeaturesUpdated(GroupFeaturesUpdated notification) {
@Override
public void onGroupFeaturesUpdated(GroupFeaturesUpdated notification) {
groupFeatures.setGroupCapabilitiesSupported(notification.getGroupCapabilitiesSupported());
groupFeatures.setGroupTypesSupported(notification.getGroupTypesSupported());
groupFeatures.setMaxGroups(notification.getMaxGroups());
groupFeatures.setGroupCapabilitiesSupported(notification.getGroupCapabilitiesSupported());
groupFeatures.setGroupTypesSupported(notification.getGroupTypesSupported());
groupFeatures.setMaxGroups(notification.getMaxGroups());
//Publish data to configuration data store
DataModificationTransaction it = this.statisticsManager.startChange();
NodeKey key = new NodeKey(notification.getId());
NodeRef ref = getNodeRef(key);
//Publish data to configuration data store
DataModificationTransaction it = this.statisticsManager.startChange();
NodeKey key = new NodeKey(notification.getId());
NodeRef ref = getNodeRef(key);
-
- final NodeBuilder nodeData = new NodeBuilder();
+
+ final NodeBuilder nodeData = new NodeBuilder();
NodeGroupFeaturesBuilder nodeGroupFeatures= new NodeGroupFeaturesBuilder();
nodeGroupFeatures.setGroupFeatures(groupFeatures.build());
NodeGroupFeaturesBuilder nodeGroupFeatures= new NodeGroupFeaturesBuilder();
nodeGroupFeatures.setGroupFeatures(groupFeatures.build());
//Update augmented data
nodeData.addAugmentation(NodeGroupFeatures.class, nodeGroupFeatures.build());
//Update augmented data
nodeData.addAugmentation(NodeGroupFeatures.class, nodeGroupFeatures.build());
InstanceIdentifier<? extends Object> refValue = ref.getValue();
it.putOperationalData(refValue, nodeData.build());
it.commit();
InstanceIdentifier<? extends Object> refValue = ref.getValue();
it.putOperationalData(refValue, nodeData.build());
it.commit();
@Override
public void onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) {
@Override
public void onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
sucLogger.debug("Received flow stats update : {}",notification.toString());
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
sucLogger.debug("Received flow stats update : {}",notification.toString());
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for(FlowAndStatisticsMapList map: notification.getFlowAndStatisticsMapList()){
short tableId = map.getTableId();
for(FlowAndStatisticsMapList map: notification.getFlowAndStatisticsMapList()){
short tableId = map.getTableId();
-
- DataModificationTransaction it = this.statisticsManager.startChange();
boolean foundOriginalFlow = false;
boolean foundOriginalFlow = false;
flow.setPriority(map.getPriority());
flow.setStrict(map.isStrict());
flow.setTableId(tableId);
flow.setPriority(map.getPriority());
flow.setStrict(map.isStrict());
flow.setTableId(tableId);
Flow flowRule = flow.build();
Flow flowRule = flow.build();
FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
stats.setByteCount(map.getByteCount());
stats.setPacketCount(map.getPacketCount());
stats.setDuration(map.getDuration());
FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
stats.setByteCount(map.getByteCount());
stats.setPacketCount(map.getPacketCount());
stats.setDuration(map.getDuration());
GenericStatistics flowStats = stats.build();
GenericStatistics flowStats = stats.build();
//Add statistics to local cache
ConcurrentMap<NodeId, NodeStatisticsAger> cache = this.statisticsManager.getStatisticsCache();
if(!cache.containsKey(notification.getId())){
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
NodeStatisticsAger nsa = cache.get(notification.getId());
//Add statistics to local cache
ConcurrentMap<NodeId, NodeStatisticsAger> cache = this.statisticsManager.getStatisticsCache();
if(!cache.containsKey(notification.getId())){
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
NodeStatisticsAger nsa = cache.get(notification.getId());
//Augment the data to the flow node
FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
//Augment the data to the flow node
FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
flowStatistics.setTableId(tableId);
flowStatisticsData.setFlowStatistics(flowStatistics.build());
flowStatistics.setTableId(tableId);
flowStatisticsData.setFlowStatistics(flowStatistics.build());
sucLogger.debug("Flow : {}",flowRule.toString());
sucLogger.debug("Statistics to augment : {}",flowStatistics.build().toString());
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
sucLogger.debug("Flow : {}",flowRule.toString());
sucLogger.debug("Statistics to augment : {}",flowStatistics.build().toString());
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
Table table= (Table)it.readConfigurationData(tableRef);
//TODO: Not a good way to do it, need to figure out better way.
Table table= (Table)it.readConfigurationData(tableRef);
//TODO: Not a good way to do it, need to figure out better way.
- //TODO: major issue in any alternate approach is that flow key is incrementally assigned
+ //TODO: major issue in any alternate approach is that flow key is incrementally assigned
//to the flows stored in data store.
// Augment same statistics to all the matching masked flow
if(table != null){
//to the flows stored in data store.
// Augment same statistics to all the matching masked flow
if(table != null){
for(Flow existingFlow : table.getFlow()){
sucLogger.debug("Existing flow in data store : {}",existingFlow.toString());
if(flowEquals(flowRule,existingFlow)){
for(Flow existingFlow : table.getFlow()){
sucLogger.debug("Existing flow in data store : {}",existingFlow.toString());
if(flowEquals(flowRule,existingFlow)){
- it = this.statisticsManager.startChange();
InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(tableId))
InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(tableId))
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Found matching flow in the datastore, augmenting statistics");
foundOriginalFlow = true;
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Found matching flow in the datastore, augmenting statistics");
foundOriginalFlow = true;
- // Update entry with timestamp of latest response
+ // Update entry with timestamp of latest response
flow.setKey(existingFlow.getKey());
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
flow.setKey(existingFlow.getKey());
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
table= (Table)it.readOperationalData(tableRef);
if(!foundOriginalFlow && table != null){
table= (Table)it.readOperationalData(tableRef);
if(!foundOriginalFlow && table != null){
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
foundOriginalFlow = true;
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
foundOriginalFlow = true;
-
- // Update entry with timestamp of latest response
+
+ // Update entry with timestamp of latest response
flow.setKey(existingFlow.getKey());
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
flow.setKey(existingFlow.getKey());
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
flowBuilder.setKey(newFlowKey);
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",flowBuilder.build());
flowBuilder.setKey(newFlowKey);
flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
sucLogger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",flowBuilder.build());
-
- // Update entry with timestamp of latest response
+
+ // Update entry with timestamp of latest response
flow.setKey(newFlowKey);
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
flow.setKey(newFlowKey);
FlowEntry flowStatsEntry = nsa.new FlowEntry(tableId,flow.build());
cache.get(notification.getId()).updateFlowStats(flowStatsEntry);
it.putOperationalData(flowRef, flowBuilder.build());
return;
NodeKey key = new NodeKey(notification.getId());
return;
NodeKey key = new NodeKey(notification.getId());
Short tableId = messageManager.getTableIdForTxId(notification.getId(),notification.getTransactionId());
if(tableId != null){
Short tableId = messageManager.getTableIdForTxId(notification.getId(),notification.getTransactionId());
if(tableId != null){
DataModificationTransaction it = this.statisticsManager.startChange();
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
DataModificationTransaction it = this.statisticsManager.startChange();
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
aggregateFlowStatisticsBuilder.setFlowCount(notification.getFlowCount());
aggregateFlowStatisticsBuilder.setPacketCount(notification.getPacketCount());
aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
aggregateFlowStatisticsBuilder.setFlowCount(notification.getFlowCount());
aggregateFlowStatisticsBuilder.setPacketCount(notification.getPacketCount());
aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
sucLogger.debug("Augment aggregate statistics: {} for table {} on Node {}",aggregateFlowStatisticsBuilder.build().toString(),tableId,key);
TableBuilder tableBuilder = new TableBuilder();
sucLogger.debug("Augment aggregate statistics: {} for table {} on Node {}",aggregateFlowStatisticsBuilder.build().toString(),tableId,key);
TableBuilder tableBuilder = new TableBuilder();
return;
NodeKey key = new NodeKey(notification.getId());
return;
NodeKey key = new NodeKey(notification.getId());
List<NodeConnectorStatisticsAndPortNumberMap> portsStats = notification.getNodeConnectorStatisticsAndPortNumberMap();
List<NodeConnectorStatisticsAndPortNumberMap> portsStats = notification.getNodeConnectorStatisticsAndPortNumberMap();
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for(NodeConnectorStatisticsAndPortNumberMap portStats : portsStats){
for(NodeConnectorStatisticsAndPortNumberMap portStats : portsStats){
-
- DataModificationTransaction it = this.statisticsManager.startChange();
- FlowCapableNodeConnectorStatisticsBuilder statisticsBuilder
+ FlowCapableNodeConnectorStatisticsBuilder statisticsBuilder
= new FlowCapableNodeConnectorStatisticsBuilder();
statisticsBuilder.setBytes(portStats.getBytes());
statisticsBuilder.setCollisionCount(portStats.getCollisionCount());
= new FlowCapableNodeConnectorStatisticsBuilder();
statisticsBuilder.setBytes(portStats.getBytes());
statisticsBuilder.setCollisionCount(portStats.getCollisionCount());
statisticsBuilder.setReceiveOverRunError(portStats.getReceiveOverRunError());
statisticsBuilder.setTransmitDrops(portStats.getTransmitDrops());
statisticsBuilder.setTransmitErrors(portStats.getTransmitErrors());
statisticsBuilder.setReceiveOverRunError(portStats.getReceiveOverRunError());
statisticsBuilder.setTransmitDrops(portStats.getTransmitDrops());
statisticsBuilder.setTransmitErrors(portStats.getTransmitErrors());
//Augment data to the node-connector
//Augment data to the node-connector
- FlowCapableNodeConnectorStatisticsDataBuilder statisticsDataBuilder =
+ FlowCapableNodeConnectorStatisticsDataBuilder statisticsDataBuilder =
new FlowCapableNodeConnectorStatisticsDataBuilder();
new FlowCapableNodeConnectorStatisticsDataBuilder();
statisticsDataBuilder.setFlowCapableNodeConnectorStatistics(statisticsBuilder.build());
statisticsDataBuilder.setFlowCapableNodeConnectorStatistics(statisticsBuilder.build());
InstanceIdentifier<NodeConnector> nodeConnectorRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key).child(NodeConnector.class, new NodeConnectorKey(portStats.getNodeConnectorId())).toInstance();
InstanceIdentifier<NodeConnector> nodeConnectorRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key).child(NodeConnector.class, new NodeConnectorKey(portStats.getNodeConnectorId())).toInstance();
NodeConnector nodeConnector = (NodeConnector)it.readOperationalData(nodeConnectorRef);
NodeConnector nodeConnector = (NodeConnector)it.readOperationalData(nodeConnectorRef);
if(nodeConnector != null){
sucLogger.debug("Augmenting port statistics {} to port {}",statisticsDataBuilder.build().toString(),nodeConnectorRef.toString());
NodeConnectorBuilder nodeConnectorBuilder = new NodeConnectorBuilder();
nodeConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, statisticsDataBuilder.build());
it.putOperationalData(nodeConnectorRef, nodeConnectorBuilder.build());
if(nodeConnector != null){
sucLogger.debug("Augmenting port statistics {} to port {}",statisticsDataBuilder.build().toString(),nodeConnectorRef.toString());
NodeConnectorBuilder nodeConnectorBuilder = new NodeConnectorBuilder();
nodeConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, statisticsDataBuilder.build());
it.putOperationalData(nodeConnectorRef, nodeConnectorBuilder.build());
return;
NodeKey key = new NodeKey(notification.getId());
return;
NodeKey key = new NodeKey(notification.getId());
List<FlowTableAndStatisticsMap> flowTablesStatsList = notification.getFlowTableAndStatisticsMap();
List<FlowTableAndStatisticsMap> flowTablesStatsList = notification.getFlowTableAndStatisticsMap();
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for (FlowTableAndStatisticsMap ftStats : flowTablesStatsList){
for (FlowTableAndStatisticsMap ftStats : flowTablesStatsList){
-
- DataModificationTransaction it = this.statisticsManager.startChange();
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(ftStats.getTableId().getValue())).toInstance();
InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(ftStats.getTableId().getValue())).toInstance();
FlowTableStatisticsDataBuilder statisticsDataBuilder = new FlowTableStatisticsDataBuilder();
FlowTableStatisticsDataBuilder statisticsDataBuilder = new FlowTableStatisticsDataBuilder();
FlowTableStatisticsBuilder statisticsBuilder = new FlowTableStatisticsBuilder();
statisticsBuilder.setActiveFlows(ftStats.getActiveFlows());
statisticsBuilder.setPacketsLookedUp(ftStats.getPacketsLookedUp());
statisticsBuilder.setPacketsMatched(ftStats.getPacketsMatched());
FlowTableStatisticsBuilder statisticsBuilder = new FlowTableStatisticsBuilder();
statisticsBuilder.setActiveFlows(ftStats.getActiveFlows());
statisticsBuilder.setPacketsLookedUp(ftStats.getPacketsLookedUp());
statisticsBuilder.setPacketsMatched(ftStats.getPacketsMatched());
statisticsDataBuilder.setFlowTableStatistics(statisticsBuilder.build());
statisticsDataBuilder.setFlowTableStatistics(statisticsBuilder.build());
sucLogger.debug("Augment flow table statistics: {} for table {} on Node {}",statisticsBuilder.build().toString(),ftStats.getTableId(),key);
sucLogger.debug("Augment flow table statistics: {} for table {} on Node {}",statisticsBuilder.build().toString(),ftStats.getTableId(),key);
TableBuilder tableBuilder = new TableBuilder();
tableBuilder.setKey(new TableKey(ftStats.getTableId().getValue()));
tableBuilder.addAugmentation(FlowTableStatisticsData.class, statisticsDataBuilder.build());
it.putOperationalData(tableRef, tableBuilder.build());
TableBuilder tableBuilder = new TableBuilder();
tableBuilder.setKey(new TableKey(ftStats.getTableId().getValue()));
tableBuilder.addAugmentation(FlowTableStatisticsData.class, statisticsDataBuilder.build());
it.putOperationalData(tableRef, tableBuilder.build());
}
@Override
public void onQueueStatisticsUpdate(QueueStatisticsUpdate notification) {
}
@Override
public void onQueueStatisticsUpdate(QueueStatisticsUpdate notification) {
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
//Check if response is for the request statistics-manager sent.
if(!messageManager.isRequestTxIdExist(notification.getId(),notification.getTransactionId(),notification.isMoreReplies()))
return;
NodeKey key = new NodeKey(notification.getId());
//Add statistics to local cache
ConcurrentMap<NodeId, NodeStatisticsAger> cache = this.statisticsManager.getStatisticsCache();
if(!cache.containsKey(notification.getId())){
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
//Add statistics to local cache
ConcurrentMap<NodeId, NodeStatisticsAger> cache = this.statisticsManager.getStatisticsCache();
if(!cache.containsKey(notification.getId())){
cache.put(notification.getId(), new NodeStatisticsAger(statisticsManager,key));
}
NodeStatisticsAger nsa = cache.get(notification.getId());
NodeStatisticsAger nsa = cache.get(notification.getId());
List<QueueIdAndStatisticsMap> queuesStats = notification.getQueueIdAndStatisticsMap();
List<QueueIdAndStatisticsMap> queuesStats = notification.getQueueIdAndStatisticsMap();
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
for(QueueIdAndStatisticsMap swQueueStats : queuesStats){
for(QueueIdAndStatisticsMap swQueueStats : queuesStats){
QueueEntry queueEntry = nsa.new QueueEntry(swQueueStats.getNodeConnectorId(),swQueueStats.getQueueId());
nsa.updateQueueStats(queueEntry);
QueueEntry queueEntry = nsa.new QueueEntry(swQueueStats.getNodeConnectorId(),swQueueStats.getQueueId());
nsa.updateQueueStats(queueEntry);
FlowCapableNodeConnectorQueueStatisticsDataBuilder queueStatisticsDataBuilder = new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
FlowCapableNodeConnectorQueueStatisticsDataBuilder queueStatisticsDataBuilder = new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
FlowCapableNodeConnectorQueueStatisticsBuilder queueStatisticsBuilder = new FlowCapableNodeConnectorQueueStatisticsBuilder();
FlowCapableNodeConnectorQueueStatisticsBuilder queueStatisticsBuilder = new FlowCapableNodeConnectorQueueStatisticsBuilder();
queueStatisticsBuilder.fieldsFrom(swQueueStats);
queueStatisticsBuilder.fieldsFrom(swQueueStats);
queueStatisticsDataBuilder.setFlowCapableNodeConnectorQueueStatistics(queueStatisticsBuilder.build());
queueStatisticsDataBuilder.setFlowCapableNodeConnectorQueueStatistics(queueStatisticsBuilder.build());
-
- DataModificationTransaction it = this.statisticsManager.startChange();
- InstanceIdentifier<Queue> queueRef
+ InstanceIdentifier<Queue> queueRef
= InstanceIdentifier.builder(Nodes.class)
.child(Node.class, key)
.child(NodeConnector.class, new NodeConnectorKey(swQueueStats.getNodeConnectorId()))
.augmentation(FlowCapableNodeConnector.class)
.child(Queue.class, new QueueKey(swQueueStats.getQueueId())).toInstance();
= InstanceIdentifier.builder(Nodes.class)
.child(Node.class, key)
.child(NodeConnector.class, new NodeConnectorKey(swQueueStats.getNodeConnectorId()))
.augmentation(FlowCapableNodeConnector.class)
.child(Queue.class, new QueueKey(swQueueStats.getQueueId())).toInstance();
QueueBuilder queueBuilder = new QueueBuilder();
queueBuilder.addAugmentation(FlowCapableNodeConnectorQueueStatisticsData.class, queueStatisticsDataBuilder.build());
queueBuilder.setKey(new QueueKey(swQueueStats.getQueueId()));
QueueBuilder queueBuilder = new QueueBuilder();
queueBuilder.addAugmentation(FlowCapableNodeConnectorQueueStatisticsData.class, queueStatisticsDataBuilder.build());
queueBuilder.setKey(new QueueKey(swQueueStats.getQueueId()));
,queueStatisticsDataBuilder.build().toString(),
swQueueStats.getQueueId(),
swQueueStats.getNodeConnectorId());
,queueStatisticsDataBuilder.build().toString(),
swQueueStats.getQueueId(),
swQueueStats.getNodeConnectorId());
it.putOperationalData(queueRef, queueBuilder.build());
it.putOperationalData(queueRef, queueBuilder.build());
}
private static NodeRef getNodeRef(NodeKey nodeKey){
InstanceIdentifierBuilder<?> builder = InstanceIdentifier.builder(Nodes.class).child(Node.class, nodeKey);
return new NodeRef(builder.toInstance());
}
}
private static NodeRef getNodeRef(NodeKey nodeKey){
InstanceIdentifierBuilder<?> builder = InstanceIdentifier.builder(Nodes.class).child(Node.class, nodeKey);
return new NodeRef(builder.toInstance());
}
-
- public boolean flowEquals(Flow statsFlow, Flow storedFlow) {
+
+ public static boolean flowEquals(Flow statsFlow, Flow storedFlow) {
if (statsFlow.getClass() != storedFlow.getClass()) {
return false;
}
if (statsFlow.getClass() != storedFlow.getClass()) {
return false;
}
/**
* Explicit equals method to compare the 'match' for flows stored in the data-stores and flow fetched from the switch.
/**
* Explicit equals method to compare the 'match' for flows stored in the data-stores and flow fetched from the switch.
- * Flow installation process has three steps
- * 1) Store flow in config data store
+ * Flow installation process has three steps
+ * 1) Store flow in config data store
* 2) and send it to plugin for installation
* 3) Flow gets installed in switch
* 2) and send it to plugin for installation
* 3) Flow gets installed in switch
- *
- * The flow user wants to install and what finally gets installed in switch can be slightly different.
+ *
+ * The flow user wants to install and what finally gets installed in switch can be slightly different.
* E.g, If user installs flow with src/dst ip=10.0.0.1/24, when it get installed in the switch
* E.g, If user installs flow with src/dst ip=10.0.0.1/24, when it get installed in the switch
- * src/dst ip will be changes to 10.0.0.0/24 because of netmask of 24. When statistics manager fetch
+ * src/dst ip will be changes to 10.0.0.0/24 because of netmask of 24. When statistics manager fetch
* stats it gets 10.0.0.0/24 rather then 10.0.0.1/24. Custom match takes care of by using masked ip
* while comparing two ip addresses.
* stats it gets 10.0.0.0/24 rather then 10.0.0.1/24. Custom match takes care of by using masked ip
* while comparing two ip addresses.
- *
- * Sometimes when user don't provide few values that is required by flow installation request, like
- * priority,hard timeout, idle timeout, cookies etc, plugin usages default values before sending
+ *
+ * Sometimes when user don't provide few values that is required by flow installation request, like
+ * priority,hard timeout, idle timeout, cookies etc, plugin usages default values before sending
* request to the switch. So when statistics manager gets flow statistics, it gets the default value.
* request to the switch. So when statistics manager gets flow statistics, it gets the default value.
- * But the flow stored in config data store don't have those defaults value. I included those checks
- * in the customer flow/match equal function.
- *
- *
+ * But the flow stored in config data store don't have those defaults value. I included those checks
+ * in the customer flow/match equal function.
+ *
+ *
* @param statsFlow
* @param storedFlow
* @return
*/
* @param statsFlow
* @param storedFlow
* @return
*/
-
- public boolean matchEquals(Match statsFlow, Match storedFlow) {
+
+ public static boolean matchEquals(Match statsFlow, Match storedFlow) {
if (statsFlow == storedFlow) {
return true;
}
if (statsFlow == storedFlow) {
return true;
}
}
protected static boolean layer3MatchEquals(Layer3Match statsLayer3Match, Layer3Match storedLayer3Match){
}
protected static boolean layer3MatchEquals(Layer3Match statsLayer3Match, Layer3Match storedLayer3Match){
- boolean verdict = true;
+ boolean verdict = true;
if(statsLayer3Match instanceof Ipv4Match && storedLayer3Match instanceof Ipv4Match){
Ipv4Match statsIpv4Match = (Ipv4Match)statsLayer3Match;
Ipv4Match storedIpv4Match = (Ipv4Match)storedLayer3Match;
if(statsLayer3Match instanceof Ipv4Match && storedLayer3Match instanceof Ipv4Match){
Ipv4Match statsIpv4Match = (Ipv4Match)statsLayer3Match;
Ipv4Match storedIpv4Match = (Ipv4Match)storedLayer3Match;
verdict = storedLayer3Match.equals(statsLayer3Match);
}
}
verdict = storedLayer3Match.equals(statsLayer3Match);
}
}
private static boolean compareNullSafe(Ipv4Prefix statsIpv4, Ipv4Prefix storedIpv4) {
boolean verdict = true;
Boolean checkDestNullValuesOut = checkNullValues(storedIpv4, statsIpv4);
private static boolean compareNullSafe(Ipv4Prefix statsIpv4, Ipv4Prefix storedIpv4) {
boolean verdict = true;
Boolean checkDestNullValuesOut = checkNullValues(storedIpv4, statsIpv4);
} else if(!IpAddressEquals(statsIpv4, storedIpv4)){
verdict = false;
}
} else if(!IpAddressEquals(statsIpv4, storedIpv4)){
verdict = false;
}
private static Boolean checkNullValues(Object v1, Object v2) {
Boolean verdict = null;
if (v1 == null && v2 != null) {
private static Boolean checkNullValues(Object v1, Object v2) {
Boolean verdict = null;
if (v1 == null && v2 != null) {
} else if (v1 == null && v2 == null) {
verdict = Boolean.TRUE;
}
} else if (v1 == null && v2 == null) {
verdict = Boolean.TRUE;
}
/**
* TODO: why don't we use the default Ipv4Prefix.equals()?
/**
* TODO: why don't we use the default Ipv4Prefix.equals()?
* @param statsIpAddress
* @param storedIpAddress
* @param statsIpAddress
* @param storedIpAddress
- * @return true if IPv4prefixes equals
+ * @return true if IPv4prefixes equals
*/
private static boolean IpAddressEquals(Ipv4Prefix statsIpAddress, Ipv4Prefix storedIpAddress) {
IntegerIpAddress statsIpAddressInt = StrIpToIntIp(statsIpAddress.getValue());
*/
private static boolean IpAddressEquals(Ipv4Prefix statsIpAddress, Ipv4Prefix storedIpAddress) {
IntegerIpAddress statsIpAddressInt = StrIpToIntIp(statsIpAddress.getValue());
private static boolean IpBasedMatch(IntegerIpAddress statsIpAddressInt,IntegerIpAddress storedIpAddressInt){
return (statsIpAddressInt.getIp() == storedIpAddressInt.getIp());
}
private static boolean IpBasedMatch(IntegerIpAddress statsIpAddressInt,IntegerIpAddress storedIpAddressInt){
return (statsIpAddressInt.getIp() == storedIpAddressInt.getIp());
}
- * Method return integer version of ip address. Converted int will be mask if
+ * Method return integer version of ip address. Converted int will be mask if
* mask specified
*/
private static IntegerIpAddress StrIpToIntIp(String ipAddresss){
* mask specified
*/
private static IntegerIpAddress StrIpToIntIp(String ipAddresss){
String[] parts = ipAddresss.split("/");
String ip = parts[0];
int prefix;
String[] parts = ipAddresss.split("/");
String ip = parts[0];
int prefix;
((addrBytes[1] & 0xFF) << 16) |
((addrBytes[2] & 0xFF) << 8) |
((addrBytes[3] & 0xFF) << 0);
((addrBytes[1] & 0xFF) << 16) |
((addrBytes[2] & 0xFF) << 8) |
((addrBytes[3] & 0xFF) << 0);
int mask = 0xffffffff << 32 - prefix;
int mask = 0xffffffff << 32 - prefix;
integerIpAddress = new IntegerIpAddress(ipInt, mask);
} catch (UnknownHostException e){
sucLogger.error("Failed to determine host IP address by name: {}", e.getMessage(), e);
integerIpAddress = new IntegerIpAddress(ipInt, mask);
} catch (UnknownHostException e){
sucLogger.error("Failed to determine host IP address by name: {}", e.getMessage(), e);
return integerIpAddress;
}
return integerIpAddress;
}
static class IntegerIpAddress{
int ip;
int mask;
static class IntegerIpAddress{
int ip;
int mask;
Code Review / controller.git / commitdiff