+ @Override
+ public void onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) {
+ NodeKey key = new NodeKey(notification.getId());
+ sucLogger.info("Received flow stats update : {}",notification.toString());
+
+ for(FlowAndStatisticsMapList map: notification.getFlowAndStatisticsMapList()){
+ short tableId = map.getTableId();
+
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
+ boolean foundOriginalFlow = false;
+
+ FlowBuilder flowBuilder = new FlowBuilder();
+
+ FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
+
+ FlowBuilder flow = new FlowBuilder();
+ flow.setContainerName(map.getContainerName());
+ flow.setBufferId(map.getBufferId());
+ flow.setCookie(map.getCookie());
+ flow.setCookieMask(map.getCookieMask());
+ flow.setFlags(map.getFlags());
+ flow.setFlowName(map.getFlowName());
+ flow.setHardTimeout(map.getHardTimeout());
+ if(map.getFlowId() != null)
+ flow.setId(new FlowId(map.getFlowId().getValue()));
+ flow.setIdleTimeout(map.getIdleTimeout());
+ flow.setInstallHw(map.isInstallHw());
+ flow.setInstructions(map.getInstructions());
+ if(map.getFlowId()!= null)
+ flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
+ flow.setMatch(map.getMatch());
+ flow.setOutGroup(map.getOutGroup());
+ flow.setOutPort(map.getOutPort());
+ flow.setPriority(map.getPriority());
+ flow.setStrict(map.isStrict());
+ flow.setTableId(tableId);
+
+ Flow flowRule = flow.build();
+
+ FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
+ stats.setByteCount(map.getByteCount());
+ stats.setPacketCount(map.getPacketCount());
+ stats.setDuration(map.getDuration());
+
+ GenericStatistics flowStats = stats.build();
+
+ //Add statistics to local cache
+ ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
+ if(!cache.containsKey(notification.getId())){
+ cache.put(notification.getId(), new NodeStatistics());
+ }
+ if(!cache.get(notification.getId()).getFlowAndStatsMap().containsKey(tableId)){
+ cache.get(notification.getId()).getFlowAndStatsMap().put(tableId, new HashMap<Flow,GenericStatistics>());
+ }
+ cache.get(notification.getId()).getFlowAndStatsMap().get(tableId).put(flowRule,flowStats);
+
+ //Augment the data to the flow node
+
+ FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
+ flowStatistics.setByteCount(flowStats.getByteCount());
+ flowStatistics.setPacketCount(flowStats.getPacketCount());
+ flowStatistics.setDuration(flowStats.getDuration());
+ flowStatistics.setContainerName(map.getContainerName());
+ flowStatistics.setBufferId(map.getBufferId());
+ flowStatistics.setCookie(map.getCookie());
+ flowStatistics.setCookieMask(map.getCookieMask());
+ flowStatistics.setFlags(map.getFlags());
+ flowStatistics.setFlowName(map.getFlowName());
+ flowStatistics.setHardTimeout(map.getHardTimeout());
+ flowStatistics.setIdleTimeout(map.getIdleTimeout());
+ flowStatistics.setInstallHw(map.isInstallHw());
+ flowStatistics.setInstructions(map.getInstructions());
+ flowStatistics.setMatch(map.getMatch());
+ flowStatistics.setOutGroup(map.getOutGroup());
+ flowStatistics.setOutPort(map.getOutPort());
+ flowStatistics.setPriority(map.getPriority());
+ flowStatistics.setStrict(map.isStrict());
+ flowStatistics.setTableId(tableId);
+
+ flowStatisticsData.setFlowStatistics(flowStatistics.build());
+
+ sucLogger.info("Flow : {}",flowRule.toString());
+ sucLogger.info("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.
+ //TODO: major issue in any alternate approach is that flow key is incrementally assigned
+ //to the flows stored in data store.
+ if(table != null){
+
+ for(Flow existingFlow : table.getFlow()){
+ sucLogger.debug("Existing flow in data store : {}",existingFlow.toString());
+ if(flowEquals(flowRule,existingFlow)){
+ InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
+ .augmentation(FlowCapableNode.class)
+ .child(Table.class, new TableKey(tableId))
+ .child(Flow.class,existingFlow.getKey()).toInstance();
+ flowBuilder.setKey(existingFlow.getKey());
+ flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
+ sucLogger.debug("Found matching flow in the datastore, augmenting statistics");
+ foundOriginalFlow = true;
+ it.putOperationalData(flowRef, flowBuilder.build());
+ it.commit();
+ break;
+ }
+ }
+ }
+
+ if(!foundOriginalFlow){
+ sucLogger.info("Associated original flow is not found in data store. Augmenting flow in operational data st");
+ //TODO: Temporary fix: format [ 0+tableid+0+unaccounted flow counter]
+ long flowKey = Long.getLong(new String("0"+Short.toString(tableId)+"0"+Integer.toString(this.unaccountedFlowsCounter)));
+ FlowKey newFlowKey = new FlowKey(new FlowId(flowKey));
+ InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
+ .augmentation(FlowCapableNode.class)
+ .child(Table.class, new TableKey(tableId))
+ .child(Flow.class,newFlowKey).toInstance();
+ flowBuilder.setKey(newFlowKey);
+ flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
+ sucLogger.debug("Flow was no present in data store, augmenting statistics as an unaccounted flow");
+ it.putOperationalData(flowRef, flowBuilder.build());
+ it.commit();
+ }
+ }
+ }
+
+ @Override
+ public void onAggregateFlowStatisticsUpdate(AggregateFlowStatisticsUpdate notification) {
+ NodeKey key = new NodeKey(notification.getId());
+ sucLogger.info("Received aggregate flow statistics update : {}",notification.toString());
+
+ Short tableId = this.statisticsManager.getMultipartMessageManager().getTableIdForTxId(notification.getTransactionId());
+ if(tableId != null){
+
+ DataModificationTransaction it = this.statisticsManager.startChange();
+
+ InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
+ .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
+
+ AggregateFlowStatisticsDataBuilder aggregateFlowStatisticsDataBuilder = new AggregateFlowStatisticsDataBuilder();
+ AggregateFlowStatisticsBuilder aggregateFlowStatisticsBuilder = new AggregateFlowStatisticsBuilder();
+ aggregateFlowStatisticsBuilder.setByteCount(notification.getByteCount());
+ aggregateFlowStatisticsBuilder.setFlowCount(notification.getFlowCount());
+ aggregateFlowStatisticsBuilder.setPacketCount(notification.getPacketCount());
+ aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
+
+ ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
+ if(!cache.containsKey(notification.getId())){
+ cache.put(notification.getId(), new NodeStatistics());
+ }
+ cache.get(notification.getId()).getTableAndAggregateFlowStatsMap().put(tableId,aggregateFlowStatisticsBuilder.build());
+
+ sucLogger.info("Augment aggregate statistics: {} for table {} on Node {}",aggregateFlowStatisticsBuilder.build().toString(),tableId,key);
+
+ TableBuilder tableBuilder = new TableBuilder();
+ tableBuilder.setKey(new TableKey(tableId));
+ tableBuilder.addAugmentation(AggregateFlowStatisticsData.class, aggregateFlowStatisticsDataBuilder.build());
+ it.putOperationalData(tableRef, tableBuilder.build());
+ it.commit();
+
+ }
+ }
+
+ @Override
+ public void onFlowStatisticsUpdated(FlowStatisticsUpdated notification) {
+ // TODO Auto-generated method stub
+ //TODO: Depricated, will clean it up once sal-compatibility is fixed.
+ //Sal-Compatibility code usage this notification event.
+
+ }
+
+ @Override
+ public void onFlowTableStatisticsUpdated(FlowTableStatisticsUpdated notification) {
+ // TODO Auto-generated method stub
+ //TODO: Need to implement it yet
+
+ }
+
+ @Override
+ public void onNodeConnectorStatisticsUpdated(NodeConnectorStatisticsUpdated notification) {
+ // TODO Auto-generated method stub
+ //TODO: Need to implement it yet
+
+ }
+
+
+
Code Review / controller.git / commitdiff