opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsUpdateCommiter.java

   1 /*
   2  * Copyright IBM Corporation, 2013.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.md.statistics.manager;
   9
  10 import java.util.HashMap;
  11 import java.util.List;
  12 import java.util.concurrent.ConcurrentMap;
  13
  14 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  15 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  16 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  17 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  18 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  19 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder;
  20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsData;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsDataBuilder;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsUpdate;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsDataBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowsStatisticsUpdate;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsListener;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.aggregate.flow.statistics.AggregateFlowStatisticsBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapListBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.statistics.FlowStatisticsBuilder;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsData;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsDataBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsUpdate;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsListener;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.statistics.FlowTableStatisticsBuilder;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueKey;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.queue.rev130925.QueueId;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.Match;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupDescStatsUpdated;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupFeaturesUpdated;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GroupStatisticsUpdated;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStats;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStatsBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeaturesBuilder;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatistics;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatisticsBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsListener;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.desc.GroupDescBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.features.GroupFeaturesBuilder;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.statistics.GroupStatisticsBuilder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.desc.stats.reply.GroupDescStats;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.statistics.reply.GroupStats;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterConfigStatsUpdated;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterFeaturesUpdated;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.MeterStatisticsUpdated;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStats;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStatsBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeaturesBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatistics;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatisticsBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsListener;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeaturesBuilder;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterConfigStatsBuilder;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatisticsBuilder;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericQueueStatistics;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericStatistics;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.NodeConnectorStatisticsUpdate;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsListener;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.flow.capable.node.connector.statistics.FlowCapableNodeConnectorStatisticsBuilder;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsData;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsDataBuilder;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsListener;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.QueueStatisticsUpdate;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.flow.capable.node.connector.queue.statistics.FlowCapableNodeConnectorQueueStatisticsBuilder;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.queue.id.and.statistics.map.QueueIdAndStatisticsMap;
 105 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 106 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
 107 import org.slf4j.Logger;
 108 import org.slf4j.LoggerFactory;
 109
 110 /**
 111  * Class implement statistics manager related listener interface and augment all the
 112  * received statistics data to data stores.
 113  * TODO: Need to add error message listener and clean-up the associated tx id
 114  * if it exists in the tx-id cache.
 115  * @author vishnoianil
 116  *
 117  */
 118 public class StatisticsUpdateCommiter implements OpendaylightGroupStatisticsListener,
 119         OpendaylightMeterStatisticsListener,
 120         OpendaylightFlowStatisticsListener,
 121         OpendaylightPortStatisticsListener,
 122         OpendaylightFlowTableStatisticsListener,
 123         OpendaylightQueueStatisticsListener{
 124
 125     public final static Logger sucLogger = LoggerFactory.getLogger(StatisticsUpdateCommiter.class);
 126
 127     private final StatisticsProvider statisticsManager;
 128
 129     private final int unaccountedFlowsCounter = 1;
 130
 131     public StatisticsUpdateCommiter(final StatisticsProvider manager){
 132
 133         this.statisticsManager = manager;
 134     }
 135
 136     public StatisticsProvider getStatisticsManager(){
 137         return statisticsManager;
 138     }
 139
 140     @Override
 141     public void onMeterConfigStatsUpdated(MeterConfigStatsUpdated notification) {
 142         //Check if response is for the request statistics-manager sent.
 143         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 144             return;
 145
 146         //Add statistics to local cache
 147         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 148         if(!cache.containsKey(notification.getId())){
 149             cache.put(notification.getId(), new NodeStatistics());
 150         }
 151         cache.get(notification.getId()).setMeterConfigStats(notification.getMeterConfigStats());
 152
 153         //Publish data to configuration data store
 154         NodeKey key = new NodeKey(notification.getId());
 155
 156         List<MeterConfigStats> eterConfigStatsList = notification.getMeterConfigStats();
 157
 158         for(MeterConfigStats meterConfigStats : eterConfigStatsList){
 159             DataModificationTransaction it = this.statisticsManager.startChange();
 160             MeterBuilder meterBuilder = new MeterBuilder();
 161             MeterKey meterKey = new MeterKey(meterConfigStats.getMeterId());
 162             meterBuilder.setKey(meterKey);
 163
 164             InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
 165                                                                                         .augmentation(FlowCapableNode.class)
 166                                                                                         .child(Meter.class,meterKey).toInstance();
 167
 168             NodeMeterConfigStatsBuilder meterConfig= new NodeMeterConfigStatsBuilder();
 169             MeterConfigStatsBuilder stats = new MeterConfigStatsBuilder();
 170             stats.fieldsFrom(meterConfigStats);
 171             meterConfig.setMeterConfigStats(stats.build());
 172
 173             //Update augmented data
 174             meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build());
 175             it.putOperationalData(meterRef, meterBuilder.build());
 176             it.commit();
 177
 178         }
 179     }
 180
 181     @Override
 182     public void onMeterStatisticsUpdated(MeterStatisticsUpdated notification) {
 183
 184         //Check if response is for the request statistics-manager sent.
 185         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 186             return;
 187
 188         //Add statistics to local cache
 189         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 190         if(!cache.containsKey(notification.getId())){
 191             cache.put(notification.getId(), new NodeStatistics());
 192         }
 193         cache.get(notification.getId()).setMeterStatistics(notification.getMeterStats());
 194
 195         NodeKey key = new NodeKey(notification.getId());
 196
 197         List<MeterStats> meterStatsList = notification.getMeterStats();
 198
 199         for(MeterStats meterStats : meterStatsList){
 200
 201             //Publish data to configuration data store
 202             DataModificationTransaction it = this.statisticsManager.startChange();
 203             MeterBuilder meterBuilder = new MeterBuilder();
 204             MeterKey meterKey = new MeterKey(meterStats.getMeterId());
 205             meterBuilder.setKey(meterKey);
 206
 207             InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
 208                                                                                         .augmentation(FlowCapableNode.class)
 209                                                                                         .child(Meter.class,meterKey).toInstance();
 210
 211             NodeMeterStatisticsBuilder meterStatsBuilder= new NodeMeterStatisticsBuilder();
 212             MeterStatisticsBuilder stats = new MeterStatisticsBuilder();
 213             stats.fieldsFrom(meterStats);
 214             meterStatsBuilder.setMeterStatistics(stats.build());
 215
 216             //Update augmented data
 217             meterBuilder.addAugmentation(NodeMeterStatistics.class, meterStatsBuilder.build());
 218             it.putOperationalData(meterRef, meterBuilder.build());
 219             it.commit();
 220         }
 221     }
 222
 223     @Override
 224     public void onGroupDescStatsUpdated(GroupDescStatsUpdated notification) {
 225
 226         //Check if response is for the request statistics-manager sent.
 227         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 228             return;
 229
 230         //Add statistics to local cache
 231         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 232         if(!cache.containsKey(notification.getId())){
 233             cache.put(notification.getId(), new NodeStatistics());
 234         }
 235         cache.get(notification.getId()).setGroupDescStats(notification.getGroupDescStats());
 236
 237         //Publish data to configuration data store
 238         NodeKey key = new NodeKey(notification.getId());
 239         List<GroupDescStats> groupDescStatsList = notification.getGroupDescStats();
 240
 241         for(GroupDescStats groupDescStats : groupDescStatsList){
 242             DataModificationTransaction it = this.statisticsManager.startChange();
 243
 244             GroupBuilder groupBuilder = new GroupBuilder();
 245             GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
 246             groupBuilder.setKey(groupKey);
 247
 248             InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
 249                                                                                         .augmentation(FlowCapableNode.class)
 250                                                                                         .child(Group.class,groupKey).toInstance();
 251
 252             NodeGroupDescStatsBuilder groupDesc= new NodeGroupDescStatsBuilder();
 253             GroupDescBuilder stats = new GroupDescBuilder();
 254             stats.fieldsFrom(groupDescStats);
 255             groupDesc.setGroupDesc(stats.build());
 256
 257             //Update augmented data
 258             groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
 259
 260             it.putOperationalData(groupRef, groupBuilder.build());
 261             it.commit();
 262         }
 263     }
 264
 265     @Override
 266     public void onGroupStatisticsUpdated(GroupStatisticsUpdated notification) {
 267
 268         //Check if response is for the request statistics-manager sent.
 269         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 270             return;
 271
 272         //Add statistics to local cache
 273         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 274         if(!cache.containsKey(notification.getId())){
 275             cache.put(notification.getId(), new NodeStatistics());
 276         }
 277         cache.get(notification.getId()).setGroupStatistics(notification.getGroupStats());
 278
 279         //Publish data to configuration data store
 280         NodeKey key = new NodeKey(notification.getId());
 281         List<GroupStats> groupStatsList = notification.getGroupStats();
 282
 283         for(GroupStats groupStats : groupStatsList){
 284             DataModificationTransaction it = this.statisticsManager.startChange();
 285
 286             GroupBuilder groupBuilder = new GroupBuilder();
 287             GroupKey groupKey = new GroupKey(groupStats.getGroupId());
 288             groupBuilder.setKey(groupKey);
 289
 290             InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class,key)
 291                                                                                         .augmentation(FlowCapableNode.class)
 292                                                                                         .child(Group.class,groupKey).toInstance();
 293
 294             NodeGroupStatisticsBuilder groupStatisticsBuilder= new NodeGroupStatisticsBuilder();
 295             GroupStatisticsBuilder stats = new GroupStatisticsBuilder();
 296             stats.fieldsFrom(groupStats);
 297             groupStatisticsBuilder.setGroupStatistics(stats.build());
 298
 299             //Update augmented data
 300             groupBuilder.addAugmentation(NodeGroupStatistics.class, groupStatisticsBuilder.build());
 301             it.putOperationalData(groupRef, groupBuilder.build());
 302             it.commit();
 303         }
 304     }
 305
 306     @Override
 307     public void onMeterFeaturesUpdated(MeterFeaturesUpdated notification) {
 308
 309         //Add statistics to local cache
 310         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 311         if(!cache.containsKey(notification.getId())){
 312             cache.put(notification.getId(), new NodeStatistics());
 313         }
 314         MeterFeaturesBuilder meterFeature = new MeterFeaturesBuilder();
 315         meterFeature.setMeterBandSupported(notification.getMeterBandSupported());
 316         meterFeature.setMeterCapabilitiesSupported(notification.getMeterCapabilitiesSupported());
 317         meterFeature.setMaxBands(notification.getMaxBands());
 318         meterFeature.setMaxColor(notification.getMaxColor());
 319         meterFeature.setMaxMeter(notification.getMaxMeter());
 320
 321         cache.get(notification.getId()).setMeterFeatures(meterFeature.build());
 322
 323         //Publish data to configuration data store
 324         DataModificationTransaction it = this.statisticsManager.startChange();
 325         NodeKey key = new NodeKey(notification.getId());
 326         NodeRef ref = getNodeRef(key);
 327
 328         final NodeBuilder nodeData = new NodeBuilder();
 329         nodeData.setKey(key);
 330
 331         NodeMeterFeaturesBuilder nodeMeterFeatures= new NodeMeterFeaturesBuilder();
 332         nodeMeterFeatures.setMeterFeatures(meterFeature.build());
 333
 334         //Update augmented data
 335         nodeData.addAugmentation(NodeMeterFeatures.class, nodeMeterFeatures.build());
 336
 337         InstanceIdentifier<? extends Object> refValue = ref.getValue();
 338         it.putOperationalData(refValue, nodeData.build());
 339         it.commit();
 340     }
 341
 342     @Override
 343     public void onGroupFeaturesUpdated(GroupFeaturesUpdated notification) {
 344
 345         //Add statistics to local cache
 346         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 347         if(!cache.containsKey(notification.getId())){
 348             cache.put(notification.getId(), new NodeStatistics());
 349         }
 350
 351         GroupFeaturesBuilder groupFeatures = new GroupFeaturesBuilder();
 352         groupFeatures.setActions(notification.getActions());
 353         groupFeatures.setGroupCapabilitiesSupported(notification.getGroupCapabilitiesSupported());
 354         groupFeatures.setGroupTypesSupported(notification.getGroupTypesSupported());
 355         groupFeatures.setMaxGroups(notification.getMaxGroups());
 356         cache.get(notification.getId()).setGroupFeatures(groupFeatures.build());
 357
 358         //Publish data to configuration data store
 359         DataModificationTransaction it = this.statisticsManager.startChange();
 360         NodeKey key = new NodeKey(notification.getId());
 361         NodeRef ref = getNodeRef(key);
 362
 363         final NodeBuilder nodeData = new NodeBuilder();
 364         nodeData.setKey(key);
 365
 366         NodeGroupFeaturesBuilder nodeGroupFeatures= new NodeGroupFeaturesBuilder();
 367         nodeGroupFeatures.setGroupFeatures(groupFeatures.build());
 368
 369         //Update augmented data
 370         nodeData.addAugmentation(NodeGroupFeatures.class, nodeGroupFeatures.build());
 371
 372         InstanceIdentifier<? extends Object> refValue = ref.getValue();
 373         it.putOperationalData(refValue, nodeData.build());
 374         it.commit();
 375     }
 376
 377     @Override
 378     public void onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) {
 379
 380         //Check if response is for the request statistics-manager sent.
 381         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 382             return;
 383
 384         NodeKey key = new NodeKey(notification.getId());
 385         sucLogger.debug("Received flow stats update : {}",notification.toString());
 386
 387         for(FlowAndStatisticsMapList map: notification.getFlowAndStatisticsMapList()){
 388             short tableId = map.getTableId();
 389
 390             DataModificationTransaction it = this.statisticsManager.startChange();
 391
 392             boolean foundOriginalFlow = false;
 393
 394             FlowBuilder flowBuilder = new FlowBuilder();
 395
 396             FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
 397
 398             FlowBuilder flow = new FlowBuilder();
 399             flow.setContainerName(map.getContainerName());
 400             flow.setBufferId(map.getBufferId());
 401             flow.setCookie(map.getCookie());
 402             flow.setCookieMask(map.getCookieMask());
 403             flow.setFlags(map.getFlags());
 404             flow.setFlowName(map.getFlowName());
 405             flow.setHardTimeout(map.getHardTimeout());
 406             if(map.getFlowId() != null)
 407                 flow.setId(new FlowId(map.getFlowId().getValue()));
 408             flow.setIdleTimeout(map.getIdleTimeout());
 409             flow.setInstallHw(map.isInstallHw());
 410             flow.setInstructions(map.getInstructions());
 411             if(map.getFlowId()!= null)
 412                 flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
 413             flow.setMatch(map.getMatch());
 414             flow.setOutGroup(map.getOutGroup());
 415             flow.setOutPort(map.getOutPort());
 416             flow.setPriority(map.getPriority());
 417             flow.setStrict(map.isStrict());
 418             flow.setTableId(tableId);
 419
 420             Flow flowRule = flow.build();
 421
 422             FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
 423             stats.setByteCount(map.getByteCount());
 424             stats.setPacketCount(map.getPacketCount());
 425             stats.setDuration(map.getDuration());
 426
 427             GenericStatistics flowStats = stats.build();
 428
 429             //Add statistics to local cache
 430             ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 431             if(!cache.containsKey(notification.getId())){
 432                 cache.put(notification.getId(), new NodeStatistics());
 433             }
 434             if(!cache.get(notification.getId()).getFlowAndStatsMap().containsKey(tableId)){
 435                 cache.get(notification.getId()).getFlowAndStatsMap().put(tableId, new HashMap<Flow,GenericStatistics>());
 436             }
 437             cache.get(notification.getId()).getFlowAndStatsMap().get(tableId).put(flowRule,flowStats);
 438
 439             //Augment the data to the flow node
 440
 441             FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
 442             flowStatistics.setByteCount(flowStats.getByteCount());
 443             flowStatistics.setPacketCount(flowStats.getPacketCount());
 444             flowStatistics.setDuration(flowStats.getDuration());
 445             flowStatistics.setContainerName(map.getContainerName());
 446             flowStatistics.setBufferId(map.getBufferId());
 447             flowStatistics.setCookie(map.getCookie());
 448             flowStatistics.setCookieMask(map.getCookieMask());
 449             flowStatistics.setFlags(map.getFlags());
 450             flowStatistics.setFlowName(map.getFlowName());
 451             flowStatistics.setHardTimeout(map.getHardTimeout());
 452             flowStatistics.setIdleTimeout(map.getIdleTimeout());
 453             flowStatistics.setInstallHw(map.isInstallHw());
 454             flowStatistics.setInstructions(map.getInstructions());
 455             flowStatistics.setMatch(map.getMatch());
 456             flowStatistics.setOutGroup(map.getOutGroup());
 457             flowStatistics.setOutPort(map.getOutPort());
 458             flowStatistics.setPriority(map.getPriority());
 459             flowStatistics.setStrict(map.isStrict());
 460             flowStatistics.setTableId(tableId);
 461
 462             flowStatisticsData.setFlowStatistics(flowStatistics.build());
 463
 464             sucLogger.debug("Flow : {}",flowRule.toString());
 465             sucLogger.debug("Statistics to augment : {}",flowStatistics.build().toString());
 466
 467             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
 468                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
 469
 470             Table table= (Table)it.readConfigurationData(tableRef);
 471
 472             //TODO: Not a good way to do it, need to figure out better way.
 473             //TODO: major issue in any alternate approach is that flow key is incrementally assigned
 474             //to the flows stored in data store.
 475             if(table != null){
 476
 477                 for(Flow existingFlow : table.getFlow()){
 478                     sucLogger.debug("Existing flow in data store : {}",existingFlow.toString());
 479                     if(flowEquals(flowRule,existingFlow)){
 480                         InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
 481                                 .augmentation(FlowCapableNode.class)
 482                                 .child(Table.class, new TableKey(tableId))
 483                                 .child(Flow.class,existingFlow.getKey()).toInstance();
 484                         flowBuilder.setKey(existingFlow.getKey());
 485                         flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 486                         sucLogger.trace("Found matching flow in the datastore, augmenting statistics");
 487                         foundOriginalFlow = true;
 488                         it.putOperationalData(flowRef, flowBuilder.build());
 489                         it.commit();
 490                         break;
 491                     }
 492                 }
 493             }
 494
 495             if(!foundOriginalFlow){
 496                 sucLogger.debug("Associated original flow is not found in data store. Augmenting flow in operational data store");
 497                 //TODO: Temporary fix: format [ 1+tableid+1+unaccounted flow counter]
 498                 long flowKey = Long.parseLong(new String("1"+Short.toString(tableId)+"1"+Integer.toString(this.unaccountedFlowsCounter)));
 499                 FlowKey newFlowKey = new FlowKey(new FlowId(Long.toString(flowKey)));
 500                 InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
 501                         .augmentation(FlowCapableNode.class)
 502                         .child(Table.class, new TableKey(tableId))
 503                         .child(Flow.class,newFlowKey).toInstance();
 504                 flowBuilder.setKey(newFlowKey);
 505                 flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 506                 sucLogger.trace("Flow was no present in data store, augmenting statistics as an unaccounted flow");
 507                 it.putOperationalData(flowRef, flowBuilder.build());
 508                 it.commit();
 509             }
 510         }
 511     }
 512
 513     @Override
 514     public void onAggregateFlowStatisticsUpdate(AggregateFlowStatisticsUpdate notification) {
 515         //Check if response is for the request statistics-manager sent.
 516         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 517             return;
 518
 519         NodeKey key = new NodeKey(notification.getId());
 520         sucLogger.debug("Received aggregate flow statistics update : {}",notification.toString());
 521
 522         Short tableId = this.statisticsManager.getMultipartMessageManager().getTableIdForTxId(notification.getTransactionId());
 523         if(tableId != null){
 524
 525             DataModificationTransaction it = this.statisticsManager.startChange();
 526
 527             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
 528                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
 529
 530             AggregateFlowStatisticsDataBuilder aggregateFlowStatisticsDataBuilder = new AggregateFlowStatisticsDataBuilder();
 531             AggregateFlowStatisticsBuilder aggregateFlowStatisticsBuilder = new AggregateFlowStatisticsBuilder();
 532             aggregateFlowStatisticsBuilder.setByteCount(notification.getByteCount());
 533             aggregateFlowStatisticsBuilder.setFlowCount(notification.getFlowCount());
 534             aggregateFlowStatisticsBuilder.setPacketCount(notification.getPacketCount());
 535             aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
 536
 537             ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 538             if(!cache.containsKey(notification.getId())){
 539                 cache.put(notification.getId(), new NodeStatistics());
 540             }
 541             cache.get(notification.getId()).getTableAndAggregateFlowStatsMap().put(tableId,aggregateFlowStatisticsBuilder.build());
 542
 543             sucLogger.debug("Augment aggregate statistics: {} for table {} on Node {}",aggregateFlowStatisticsBuilder.build().toString(),tableId,key);
 544
 545             TableBuilder tableBuilder = new TableBuilder();
 546             tableBuilder.setKey(new TableKey(tableId));
 547             tableBuilder.addAugmentation(AggregateFlowStatisticsData.class, aggregateFlowStatisticsDataBuilder.build());
 548             it.putOperationalData(tableRef, tableBuilder.build());
 549             it.commit();
 550
 551         }
 552     }
 553
 554     @Override
 555     public void onNodeConnectorStatisticsUpdate(NodeConnectorStatisticsUpdate notification) {
 556         //Check if response is for the request statistics-manager sent.
 557         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 558             return;
 559
 560         NodeKey key = new NodeKey(notification.getId());
 561         sucLogger.debug("Received port stats update : {}",notification.toString());
 562
 563         //Add statistics to local cache
 564         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 565         if(!cache.containsKey(notification.getId())){
 566             cache.put(notification.getId(), new NodeStatistics());
 567         }
 568
 569
 570         List<NodeConnectorStatisticsAndPortNumberMap> portsStats = notification.getNodeConnectorStatisticsAndPortNumberMap();
 571         for(NodeConnectorStatisticsAndPortNumberMap portStats : portsStats){
 572
 573             DataModificationTransaction it = this.statisticsManager.startChange();
 574
 575             FlowCapableNodeConnectorStatisticsBuilder statisticsBuilder
 576                                             = new FlowCapableNodeConnectorStatisticsBuilder();
 577             statisticsBuilder.setBytes(portStats.getBytes());
 578             statisticsBuilder.setCollisionCount(portStats.getCollisionCount());
 579             statisticsBuilder.setDuration(portStats.getDuration());
 580             statisticsBuilder.setPackets(portStats.getPackets());
 581             statisticsBuilder.setReceiveCrcError(portStats.getReceiveCrcError());
 582             statisticsBuilder.setReceiveDrops(portStats.getReceiveDrops());
 583             statisticsBuilder.setReceiveErrors(portStats.getReceiveErrors());
 584             statisticsBuilder.setReceiveFrameError(portStats.getReceiveFrameError());
 585             statisticsBuilder.setReceiveOverRunError(portStats.getReceiveOverRunError());
 586             statisticsBuilder.setTransmitDrops(portStats.getTransmitDrops());
 587             statisticsBuilder.setTransmitErrors(portStats.getTransmitErrors());
 588
 589             //Update data in the cache
 590             cache.get(notification.getId()).getNodeConnectorStats().put(portStats.getNodeConnectorId(), statisticsBuilder.build());
 591
 592             //Augment data to the node-connector
 593             FlowCapableNodeConnectorStatisticsDataBuilder statisticsDataBuilder =
 594                     new FlowCapableNodeConnectorStatisticsDataBuilder();
 595
 596             statisticsDataBuilder.setFlowCapableNodeConnectorStatistics(statisticsBuilder.build());
 597
 598             InstanceIdentifier<NodeConnector> nodeConnectorRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key).child(NodeConnector.class, new NodeConnectorKey(portStats.getNodeConnectorId())).toInstance();
 599
 600             NodeConnector nodeConnector = (NodeConnector)it.readOperationalData(nodeConnectorRef);
 601
 602             if(nodeConnector != null){
 603                 sucLogger.debug("Augmenting port statistics {} to port {}",statisticsDataBuilder.build().toString(),nodeConnectorRef.toString());
 604                 NodeConnectorBuilder nodeConnectorBuilder = new NodeConnectorBuilder();
 605                 nodeConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, statisticsDataBuilder.build());
 606                 it.putOperationalData(nodeConnectorRef, nodeConnectorBuilder.build());
 607                 it.commit();
 608             }
 609         }
 610     }
 611
 612     @Override
 613     public void onFlowTableStatisticsUpdate(FlowTableStatisticsUpdate notification) {
 614         //Check if response is for the request statistics-manager sent.
 615         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 616             return;
 617
 618         NodeKey key = new NodeKey(notification.getId());
 619         sucLogger.debug("Received flow table statistics update : {}",notification.toString());
 620
 621         List<FlowTableAndStatisticsMap> flowTablesStatsList = notification.getFlowTableAndStatisticsMap();
 622         for (FlowTableAndStatisticsMap ftStats : flowTablesStatsList){
 623
 624             DataModificationTransaction it = this.statisticsManager.startChange();
 625
 626             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, key)
 627                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(ftStats.getTableId().getValue())).toInstance();
 628
 629             FlowTableStatisticsDataBuilder statisticsDataBuilder = new FlowTableStatisticsDataBuilder();
 630
 631             FlowTableStatisticsBuilder statisticsBuilder = new FlowTableStatisticsBuilder();
 632             statisticsBuilder.setActiveFlows(ftStats.getActiveFlows());
 633             statisticsBuilder.setPacketsLookedUp(ftStats.getPacketsLookedUp());
 634             statisticsBuilder.setPacketsMatched(ftStats.getPacketsMatched());
 635
 636             statisticsDataBuilder.setFlowTableStatistics(statisticsBuilder.build());
 637
 638             ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 639             if(!cache.containsKey(notification.getId())){
 640                 cache.put(notification.getId(), new NodeStatistics());
 641             }
 642             cache.get(notification.getId()).getFlowTableAndStatisticsMap().put(ftStats.getTableId().getValue(),statisticsBuilder.build());
 643
 644             sucLogger.debug("Augment flow table statistics: {} for table {} on Node {}",statisticsBuilder.build().toString(),ftStats.getTableId(),key);
 645
 646             TableBuilder tableBuilder = new TableBuilder();
 647             tableBuilder.setKey(new TableKey(ftStats.getTableId().getValue()));
 648             tableBuilder.addAugmentation(FlowTableStatisticsData.class, statisticsDataBuilder.build());
 649             it.putOperationalData(tableRef, tableBuilder.build());
 650             it.commit();
 651         }
 652     }
 653
 654     @Override
 655     public void onQueueStatisticsUpdate(QueueStatisticsUpdate notification) {
 656
 657         //Check if response is for the request statistics-manager sent.
 658         if(this.statisticsManager.getMultipartMessageManager().removeTxId(notification.getTransactionId()) == null)
 659             return;
 660
 661         NodeKey key = new NodeKey(notification.getId());
 662         sucLogger.debug("Received queue stats update : {}",notification.toString());
 663
 664         //Add statistics to local cache
 665         ConcurrentMap<NodeId, NodeStatistics> cache = this.statisticsManager.getStatisticsCache();
 666         if(!cache.containsKey(notification.getId())){
 667             cache.put(notification.getId(), new NodeStatistics());
 668         }
 669
 670         List<QueueIdAndStatisticsMap> queuesStats = notification.getQueueIdAndStatisticsMap();
 671         for(QueueIdAndStatisticsMap swQueueStats : queuesStats){
 672
 673             if(!cache.get(notification.getId()).getNodeConnectorAndQueuesStatsMap().containsKey(swQueueStats.getNodeConnectorId())){
 674                 cache.get(notification.getId()).getNodeConnectorAndQueuesStatsMap().put(swQueueStats.getNodeConnectorId(), new HashMap<QueueId,GenericQueueStatistics>());
 675             }
 676
 677             FlowCapableNodeConnectorQueueStatisticsDataBuilder queueStatisticsDataBuilder = new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
 678
 679             FlowCapableNodeConnectorQueueStatisticsBuilder queueStatisticsBuilder = new FlowCapableNodeConnectorQueueStatisticsBuilder();
 680
 681             queueStatisticsBuilder.fieldsFrom(swQueueStats);
 682
 683             queueStatisticsDataBuilder.setFlowCapableNodeConnectorQueueStatistics(queueStatisticsBuilder.build());
 684
 685             cache.get(notification.getId()).getNodeConnectorAndQueuesStatsMap()
 686                                             .get(swQueueStats.getNodeConnectorId())
 687                                             .put(swQueueStats.getQueueId(), queueStatisticsBuilder.build());
 688
 689
 690             DataModificationTransaction it = this.statisticsManager.startChange();
 691
 692             InstanceIdentifier<Queue> queueRef
 693                     = InstanceIdentifier.builder(Nodes.class)
 694                                         .child(Node.class, key)
 695                                         .child(NodeConnector.class, new NodeConnectorKey(swQueueStats.getNodeConnectorId()))
 696                                         .augmentation(FlowCapableNodeConnector.class)
 697                                         .child(Queue.class, new QueueKey(swQueueStats.getQueueId())).toInstance();
 698
 699             QueueBuilder queueBuilder = new QueueBuilder();
 700             queueBuilder.addAugmentation(FlowCapableNodeConnectorQueueStatisticsData.class, queueStatisticsDataBuilder.build());
 701             queueBuilder.setKey(new QueueKey(swQueueStats.getQueueId()));
 702
 703             sucLogger.trace("Augmenting queue statistics {} of queue {} to port {}"
 704                                         ,queueStatisticsDataBuilder.build().toString(),
 705                                         swQueueStats.getQueueId(),
 706                                         swQueueStats.getNodeConnectorId());
 707
 708             it.putOperationalData(queueRef, queueBuilder.build());
 709             it.commit();
 710
 711         }
 712
 713     }
 714
 715     private NodeRef getNodeRef(NodeKey nodeKey){
 716         InstanceIdentifierBuilder<?> builder = InstanceIdentifier.builder(Nodes.class).child(Node.class, nodeKey);
 717         return new NodeRef(builder.toInstance());
 718     }
 719
 720     public boolean flowEquals(Flow statsFlow, Flow storedFlow) {
 721         if (statsFlow.getClass() != storedFlow.getClass()) {
 722             return false;
 723         }
 724         if (statsFlow.getBufferId()== null) {
 725             if (storedFlow.getBufferId() != null) {
 726                 return false;
 727             }
 728         } else if(!statsFlow.getBufferId().equals(storedFlow.getBufferId())) {
 729             return false;
 730         }
 731         if (statsFlow.getContainerName()== null) {
 732             if (storedFlow.getContainerName()!= null) {
 733                 return false;
 734             }
 735         } else if(!statsFlow.getContainerName().equals(storedFlow.getContainerName())) {
 736             return false;
 737         }
 738         if (statsFlow.getCookie()== null) {
 739             if (storedFlow.getCookie()!= null) {
 740                 return false;
 741             }
 742         } else if(!statsFlow.getCookie().equals(storedFlow.getCookie())) {
 743             return false;
 744         }
 745         if (statsFlow.getMatch()== null) {
 746             if (storedFlow.getMatch() != null) {
 747                 return false;
 748             }
 749         } //else if(!statsFlow.getMatch().equals(storedFlow.getMatch())) {
 750         else if(!matchEquals(statsFlow.getMatch(), storedFlow.getMatch())) {
 751             return false;
 752         }
 753         if (statsFlow.getHardTimeout() == null) {
 754             if (storedFlow.getHardTimeout() != null) {
 755                 return false;
 756             }
 757         } else if(!statsFlow.getHardTimeout().equals(storedFlow.getHardTimeout() )) {
 758             return false;
 759         }
 760         if (statsFlow.getIdleTimeout()== null) {
 761             if (storedFlow.getIdleTimeout() != null) {
 762                 return false;
 763             }
 764         } else if(!statsFlow.getIdleTimeout().equals(storedFlow.getIdleTimeout())) {
 765             return false;
 766         }
 767         if (statsFlow.getPriority() == null) {
 768             if (storedFlow.getPriority() != null) {
 769                 return false;
 770             }
 771         } else if(!statsFlow.getPriority().equals(storedFlow.getPriority())) {
 772             return false;
 773         }
 774         if (statsFlow.getTableId() == null) {
 775             if (storedFlow.getTableId() != null) {
 776                 return false;
 777             }
 778         } else if(!statsFlow.getTableId().equals(storedFlow.getTableId())) {
 779             return false;
 780         }
 781         return true;
 782     }
 783
 784     /**
 785      * Explicit equals method to compare the 'match' for flows stored in the data-stores and flow fetched from the switch.
 786      * Usecase: e.g If user don't set any ethernet source and destination address for match,data store will store null for
 787      * these address.
 788      * e.g [_ethernetMatch=EthernetMatch [_ethernetDestination=null, _ethernetSource=null, _ethernetType=
 789      * EthernetType [_type=EtherType [_value=2048], _mask=null, augmentation=[]]
 790      *
 791      * But when you fetch the flows from switch, openflow driver library converts all zero bytes of mac address in the
 792      * message stream to 00:00:00:00:00:00. Following string shows how library interpret the zero mac address bytes and
 793      * eventually when translator convert it to MD-SAL match, this is how it looks
 794      * [_ethernetDestination=EthernetDestination [_address=MacAddress [_value=00:00:00:00:00:00], _mask=null, augmentation=[]],
 795      * _ethernetSource=EthernetSource [_address=MacAddress [_value=00:00:00:00:00:00], _mask=null, augmentation=[]],
 796      * _ethernetType=EthernetType [_type=EtherType [_value=2048], _mask=null, augmentation=[]]
 797      *
 798      * Similarly for inPort, if user/application don't set any value for it, FRM will store null value for it in data store.
 799      * When we fetch the same flow (with its statistics) from switch, plugin converts its value to openflow:X:0.
 800      *  e.g _inPort=Uri [_value=openflow:1:0]
 801      *
 802      * So this custom equals method add additional check to take care of these scenario, in case any match element is null in data-store-flow, but not
 803      * in the flow fetched from switch.
 804      *
 805      * @param statsFlow
 806      * @param storedFlow
 807      * @return
 808      */
 809
 810     public boolean matchEquals(Match statsFlow, Match storedFlow) {
 811         if (statsFlow == storedFlow) {
 812             return true;
 813         }
 814         if (storedFlow.getClass() != statsFlow.getClass()) {
 815             return false;
 816         }
 817         if (storedFlow.getEthernetMatch() == null) {
 818             if (statsFlow.getEthernetMatch() != null) {
 819                 return false;
 820             }
 821         } else if(!storedFlow.getEthernetMatch().equals(statsFlow.getEthernetMatch())) {
 822             return false;
 823         }
 824         if (storedFlow.getIcmpv4Match()== null) {
 825             if (statsFlow.getIcmpv4Match() != null) {
 826                 return false;
 827             }
 828         } else if(!storedFlow.getIcmpv4Match().equals(statsFlow.getIcmpv4Match())) {
 829             return false;
 830         }
 831         if (storedFlow.getIcmpv6Match() == null) {
 832             if (statsFlow.getIcmpv6Match() != null) {
 833                 return false;
 834             }
 835         } else if(!storedFlow.getIcmpv6Match().equals(statsFlow.getIcmpv6Match())) {
 836             return false;
 837         }
 838         if (storedFlow.getInPhyPort() == null) {
 839             if (statsFlow.getInPhyPort() != null) {
 840                 return false;
 841             }
 842         } else if(!storedFlow.getInPhyPort().equals(statsFlow.getInPhyPort())) {
 843             return false;
 844         }
 845         if (storedFlow.getInPort()== null) {
 846             if (statsFlow.getInPort() != null) {
 847                 return false;
 848             }
 849         } else if(!storedFlow.getInPort().equals(statsFlow.getInPort())) {
 850             return false;
 851         }
 852         if (storedFlow.getIpMatch()== null) {
 853             if (statsFlow.getIpMatch() != null) {
 854                 return false;
 855             }
 856         } else if(!storedFlow.getIpMatch().equals(statsFlow.getIpMatch())) {
 857             return false;
 858         }
 859         if (storedFlow.getLayer3Match()== null) {
 860             if (statsFlow.getLayer3Match() != null) {
 861                     return false;
 862             }
 863         } else if(!storedFlow.getLayer3Match().equals(statsFlow.getLayer3Match())) {
 864             return false;
 865         }
 866         if (storedFlow.getLayer4Match()== null) {
 867             if (statsFlow.getLayer4Match() != null) {
 868                 return false;
 869             }
 870         } else if(!storedFlow.getLayer4Match().equals(statsFlow.getLayer4Match())) {
 871             return false;
 872         }
 873         if (storedFlow.getMetadata() == null) {
 874             if (statsFlow.getMetadata() != null) {
 875                 return false;
 876             }
 877         } else if(!storedFlow.getMetadata().equals(statsFlow.getMetadata())) {
 878             return false;
 879         }
 880         if (storedFlow.getProtocolMatchFields() == null) {
 881             if (statsFlow.getProtocolMatchFields() != null) {
 882                 return false;
 883             }
 884         } else if(!storedFlow.getProtocolMatchFields().equals(statsFlow.getProtocolMatchFields())) {
 885             return false;
 886         }
 887         if (storedFlow.getTunnel()== null) {
 888             if (statsFlow.getTunnel() != null) {
 889                 return false;
 890             }
 891         } else if(!storedFlow.getTunnel().equals(statsFlow.getTunnel())) {
 892             return false;
 893         }
 894         if (storedFlow.getVlanMatch()== null) {
 895             if (statsFlow.getVlanMatch() != null) {
 896                 return false;
 897             }
 898         } else if(!storedFlow.getVlanMatch().equals(statsFlow.getVlanMatch())) {
 899             return false;
 900         }
 901         return true;
 902     }
 903 }