opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/NodeStatisticsHandler.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.Iterator;
  12 import java.util.List;
  13 import java.util.Map;
  14 import java.util.Map.Entry;
  15 import java.util.concurrent.TimeUnit;
  16
  17 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  18 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  19 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsData;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsDataBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsDataBuilder;
  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.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.statistics.FlowTableStatistics;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.statistics.FlowTableStatisticsBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueKey;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.queue.rev130925.QueueId;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStats;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupDescStatsBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeaturesBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatistics;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupStatisticsBuilder;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.desc.GroupDescBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.features.GroupFeaturesBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.statistics.GroupStatisticsBuilder;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupFeatures;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.desc.stats.reply.GroupDescStats;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.statistics.reply.GroupStats;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeBuilder;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStats;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStatsBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeaturesBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatistics;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatisticsBuilder;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeaturesBuilder;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterConfigStatsBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatisticsBuilder;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterFeatures;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.AggregateFlowStatistics;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericStatistics;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.flow.capable.node.connector.statistics.FlowCapableNodeConnectorStatisticsBuilder;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsData;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsDataBuilder;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.flow.capable.node.connector.queue.statistics.FlowCapableNodeConnectorQueueStatisticsBuilder;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.queue.id.and.statistics.map.QueueIdAndStatisticsMap;
  92 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  93 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
  94 import org.slf4j.Logger;
  95 import org.slf4j.LoggerFactory;
  96
  97 import com.google.common.base.Preconditions;
  98
  99 /**
 100  * This class handles the lifecycle of per-node statistics. It receives data
 101  * from StatisticsListener, stores it in the data store and keeps track of
 102  * when the data should be removed.
 103  *
 104  * @author avishnoi@in.ibm.com
 105  */
 106 public class NodeStatisticsHandler implements AutoCloseable {
 107     private static final Logger logger = LoggerFactory.getLogger(NodeStatisticsHandler.class);
 108     private static final int NUMBER_OF_WAIT_CYCLES = 2;
 109
 110     private final Map<GroupDescStats,Long> groupDescStatsUpdate = new HashMap<>();
 111     private final Map<MeterConfigStats,Long> meterConfigStatsUpdate = new HashMap<>();
 112     private final Map<FlowEntry,Long> flowStatsUpdate = new HashMap<>();
 113     private final Map<QueueEntry,Long> queuesStatsUpdate = new HashMap<>();
 114     private final InstanceIdentifier<Node> targetNodeIdentifier;
 115     private final StatisticsProvider statisticsProvider;
 116     private final NodeKey targetNodeKey;
 117     private int unaccountedFlowsCounter = 1;
 118
 119     public NodeStatisticsHandler(StatisticsProvider statisticsProvider, NodeKey nodeKey){
 120         this.statisticsProvider = Preconditions.checkNotNull(statisticsProvider);
 121         this.targetNodeKey = Preconditions.checkNotNull(nodeKey);
 122         this.targetNodeIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).build();
 123     }
 124
 125     private static class FlowEntry {
 126         private final Short tableId;
 127         private final Flow flow;
 128
 129         public FlowEntry(Short tableId, Flow flow){
 130             this.tableId = tableId;
 131             this.flow = flow;
 132         }
 133
 134         public Short getTableId() {
 135             return tableId;
 136         }
 137
 138         public Flow getFlow() {
 139             return flow;
 140         }
 141
 142         @Override
 143         public int hashCode() {
 144             final int prime = 31;
 145             int result = 1;
 146             result = prime * result + ((flow == null) ? 0 : flow.hashCode());
 147             result = prime * result + ((tableId == null) ? 0 : tableId.hashCode());
 148             return result;
 149         }
 150
 151         @Override
 152         public boolean equals(Object obj) {
 153             if (this == obj)
 154                 return true;
 155             if (obj == null)
 156                 return false;
 157             if (getClass() != obj.getClass())
 158                 return false;
 159             FlowEntry other = (FlowEntry) obj;
 160             if (flow == null) {
 161                 if (other.flow != null)
 162                     return false;
 163             } else if (!flow.equals(other.flow))
 164                 return false;
 165             if (tableId == null) {
 166                 if (other.tableId != null)
 167                     return false;
 168             } else if (!tableId.equals(other.tableId))
 169                 return false;
 170             return true;
 171         }
 172     }
 173
 174     private static final class QueueEntry{
 175         private final NodeConnectorId nodeConnectorId;
 176         private final QueueId queueId;
 177         public QueueEntry(NodeConnectorId ncId, QueueId queueId){
 178             this.nodeConnectorId = ncId;
 179             this.queueId = queueId;
 180         }
 181         public NodeConnectorId getNodeConnectorId() {
 182             return nodeConnectorId;
 183         }
 184         public QueueId getQueueId() {
 185             return queueId;
 186         }
 187         @Override
 188         public int hashCode() {
 189             final int prime = 31;
 190             int result = 1;
 191             result = prime * result + ((nodeConnectorId == null) ? 0 : nodeConnectorId.hashCode());
 192             result = prime * result + ((queueId == null) ? 0 : queueId.hashCode());
 193             return result;
 194         }
 195         @Override
 196         public boolean equals(Object obj) {
 197             if (this == obj) {
 198                 return true;
 199             }
 200             if (obj == null) {
 201                 return false;
 202             }
 203             if (!(obj instanceof QueueEntry)) {
 204                 return false;
 205             }
 206             QueueEntry other = (QueueEntry) obj;
 207             if (nodeConnectorId == null) {
 208                 if (other.nodeConnectorId != null) {
 209                     return false;
 210                 }
 211             } else if (!nodeConnectorId.equals(other.nodeConnectorId)) {
 212                 return false;
 213             }
 214             if (queueId == null) {
 215                 if (other.queueId != null) {
 216                     return false;
 217                 }
 218             } else if (!queueId.equals(other.queueId)) {
 219                 return false;
 220             }
 221             return true;
 222         }
 223     }
 224
 225     public NodeKey getTargetNodeKey() {
 226         return targetNodeKey;
 227     }
 228
 229     public synchronized void updateGroupDescStats(List<GroupDescStats> list){
 230         final Long expiryTime = getExpiryTime();
 231         final DataModificationTransaction trans = statisticsProvider.startChange();
 232
 233         for (GroupDescStats groupDescStats : list) {
 234             GroupBuilder groupBuilder = new GroupBuilder();
 235             GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
 236             groupBuilder.setKey(groupKey);
 237
 238             InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 239                                                                                         .augmentation(FlowCapableNode.class)
 240                                                                                         .child(Group.class,groupKey).toInstance();
 241
 242             NodeGroupDescStatsBuilder groupDesc= new NodeGroupDescStatsBuilder();
 243             GroupDescBuilder stats = new GroupDescBuilder();
 244             stats.fieldsFrom(groupDescStats);
 245             groupDesc.setGroupDesc(stats.build());
 246
 247             //Update augmented data
 248             groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
 249
 250             trans.putOperationalData(groupRef, groupBuilder.build());
 251             this.groupDescStatsUpdate.put(groupDescStats, expiryTime);
 252         }
 253
 254         trans.commit();
 255     }
 256
 257
 258     public synchronized void updateGroupStats(List<GroupStats> list) {
 259         final DataModificationTransaction trans = statisticsProvider.startChange();
 260
 261         for(GroupStats groupStats : list) {
 262             GroupBuilder groupBuilder = new GroupBuilder();
 263             GroupKey groupKey = new GroupKey(groupStats.getGroupId());
 264             groupBuilder.setKey(groupKey);
 265
 266             InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 267                                                                                         .augmentation(FlowCapableNode.class)
 268                                                                                         .child(Group.class,groupKey).toInstance();
 269
 270             NodeGroupStatisticsBuilder groupStatisticsBuilder= new NodeGroupStatisticsBuilder();
 271             GroupStatisticsBuilder stats = new GroupStatisticsBuilder();
 272             stats.fieldsFrom(groupStats);
 273             groupStatisticsBuilder.setGroupStatistics(stats.build());
 274
 275             //Update augmented data
 276             groupBuilder.addAugmentation(NodeGroupStatistics.class, groupStatisticsBuilder.build());
 277             trans.putOperationalData(groupRef, groupBuilder.build());
 278
 279             // FIXME: should we be tracking this data?
 280         }
 281
 282         trans.commit();
 283     }
 284
 285     public synchronized void updateMeterConfigStats(List<MeterConfigStats> list) {
 286         final Long expiryTime = getExpiryTime();
 287         final DataModificationTransaction trans = statisticsProvider.startChange();
 288
 289         for(MeterConfigStats meterConfigStats : list) {
 290             MeterBuilder meterBuilder = new MeterBuilder();
 291             MeterKey meterKey = new MeterKey(meterConfigStats.getMeterId());
 292             meterBuilder.setKey(meterKey);
 293
 294             InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 295                                                                                         .augmentation(FlowCapableNode.class)
 296                                                                                         .child(Meter.class,meterKey).toInstance();
 297
 298             NodeMeterConfigStatsBuilder meterConfig= new NodeMeterConfigStatsBuilder();
 299             MeterConfigStatsBuilder stats = new MeterConfigStatsBuilder();
 300             stats.fieldsFrom(meterConfigStats);
 301             meterConfig.setMeterConfigStats(stats.build());
 302
 303             //Update augmented data
 304             meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build());
 305
 306             trans.putOperationalData(meterRef, meterBuilder.build());
 307             this.meterConfigStatsUpdate.put(meterConfigStats, expiryTime);
 308         }
 309
 310         trans.commit();
 311     }
 312
 313
 314     public synchronized void updateMeterStats(List<MeterStats> list) {
 315         final DataModificationTransaction trans = statisticsProvider.startChange();
 316
 317         for(MeterStats meterStats : list) {
 318             MeterBuilder meterBuilder = new MeterBuilder();
 319             MeterKey meterKey = new MeterKey(meterStats.getMeterId());
 320             meterBuilder.setKey(meterKey);
 321
 322             InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 323                                                                                         .augmentation(FlowCapableNode.class)
 324                                                                                         .child(Meter.class,meterKey).toInstance();
 325
 326             NodeMeterStatisticsBuilder meterStatsBuilder= new NodeMeterStatisticsBuilder();
 327             MeterStatisticsBuilder stats = new MeterStatisticsBuilder();
 328             stats.fieldsFrom(meterStats);
 329             meterStatsBuilder.setMeterStatistics(stats.build());
 330
 331             //Update augmented data
 332             meterBuilder.addAugmentation(NodeMeterStatistics.class, meterStatsBuilder.build());
 333             trans.putOperationalData(meterRef, meterBuilder.build());
 334
 335             // FIXME: should we be tracking this data?
 336         }
 337
 338         trans.commit();
 339     }
 340
 341     public synchronized void updateQueueStats(List<QueueIdAndStatisticsMap> list) {
 342         final Long expiryTime = getExpiryTime();
 343         final DataModificationTransaction trans = statisticsProvider.startChange();
 344
 345         for (QueueIdAndStatisticsMap swQueueStats : list) {
 346
 347             QueueEntry queueEntry = new QueueEntry(swQueueStats.getNodeConnectorId(),swQueueStats.getQueueId());
 348
 349             FlowCapableNodeConnectorQueueStatisticsDataBuilder queueStatisticsDataBuilder = new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
 350
 351             FlowCapableNodeConnectorQueueStatisticsBuilder queueStatisticsBuilder = new FlowCapableNodeConnectorQueueStatisticsBuilder();
 352
 353             queueStatisticsBuilder.fieldsFrom(swQueueStats);
 354
 355             queueStatisticsDataBuilder.setFlowCapableNodeConnectorQueueStatistics(queueStatisticsBuilder.build());
 356
 357             InstanceIdentifier<Queue> queueRef
 358                     = InstanceIdentifier.builder(Nodes.class)
 359                                         .child(Node.class, targetNodeKey)
 360                                         .child(NodeConnector.class, new NodeConnectorKey(swQueueStats.getNodeConnectorId()))
 361                                         .augmentation(FlowCapableNodeConnector.class)
 362                                         .child(Queue.class, new QueueKey(swQueueStats.getQueueId())).toInstance();
 363
 364             QueueBuilder queueBuilder = new QueueBuilder();
 365             FlowCapableNodeConnectorQueueStatisticsData qsd = queueStatisticsDataBuilder.build();
 366             queueBuilder.addAugmentation(FlowCapableNodeConnectorQueueStatisticsData.class, qsd);
 367             queueBuilder.setKey(new QueueKey(swQueueStats.getQueueId()));
 368
 369             logger.debug("Augmenting queue statistics {} of queue {} to port {}",
 370                                         qsd,
 371                                         swQueueStats.getQueueId(),
 372                                         swQueueStats.getNodeConnectorId());
 373
 374             trans.putOperationalData(queueRef, queueBuilder.build());
 375             this.queuesStatsUpdate.put(queueEntry, expiryTime);
 376         }
 377
 378         trans.commit();
 379     }
 380
 381     public synchronized void updateFlowTableStats(List<FlowTableAndStatisticsMap> list) {
 382         final DataModificationTransaction trans = statisticsProvider.startChange();
 383
 384         for (FlowTableAndStatisticsMap ftStats : list) {
 385
 386             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 387                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(ftStats.getTableId().getValue())).toInstance();
 388
 389             FlowTableStatisticsDataBuilder statisticsDataBuilder = new FlowTableStatisticsDataBuilder();
 390
 391             FlowTableStatisticsBuilder statisticsBuilder = new FlowTableStatisticsBuilder();
 392             statisticsBuilder.setActiveFlows(ftStats.getActiveFlows());
 393             statisticsBuilder.setPacketsLookedUp(ftStats.getPacketsLookedUp());
 394             statisticsBuilder.setPacketsMatched(ftStats.getPacketsMatched());
 395
 396             final FlowTableStatistics stats = statisticsBuilder.build();
 397             statisticsDataBuilder.setFlowTableStatistics(stats);
 398
 399             logger.debug("Augment flow table statistics: {} for table {} on Node {}",
 400                     stats,ftStats.getTableId(), targetNodeKey);
 401
 402             TableBuilder tableBuilder = new TableBuilder();
 403             tableBuilder.setKey(new TableKey(ftStats.getTableId().getValue()));
 404             tableBuilder.addAugmentation(FlowTableStatisticsData.class, statisticsDataBuilder.build());
 405             trans.putOperationalData(tableRef, tableBuilder.build());
 406
 407             // FIXME: should we be tracking this data?
 408         }
 409
 410         trans.commit();
 411     }
 412
 413     public synchronized void updateNodeConnectorStats(List<NodeConnectorStatisticsAndPortNumberMap> list) {
 414         final DataModificationTransaction trans = statisticsProvider.startChange();
 415
 416         for(NodeConnectorStatisticsAndPortNumberMap portStats : list) {
 417
 418             FlowCapableNodeConnectorStatisticsBuilder statisticsBuilder
 419                                             = new FlowCapableNodeConnectorStatisticsBuilder();
 420             statisticsBuilder.setBytes(portStats.getBytes());
 421             statisticsBuilder.setCollisionCount(portStats.getCollisionCount());
 422             statisticsBuilder.setDuration(portStats.getDuration());
 423             statisticsBuilder.setPackets(portStats.getPackets());
 424             statisticsBuilder.setReceiveCrcError(portStats.getReceiveCrcError());
 425             statisticsBuilder.setReceiveDrops(portStats.getReceiveDrops());
 426             statisticsBuilder.setReceiveErrors(portStats.getReceiveErrors());
 427             statisticsBuilder.setReceiveFrameError(portStats.getReceiveFrameError());
 428             statisticsBuilder.setReceiveOverRunError(portStats.getReceiveOverRunError());
 429             statisticsBuilder.setTransmitDrops(portStats.getTransmitDrops());
 430             statisticsBuilder.setTransmitErrors(portStats.getTransmitErrors());
 431
 432             //Augment data to the node-connector
 433             FlowCapableNodeConnectorStatisticsDataBuilder statisticsDataBuilder =
 434                     new FlowCapableNodeConnectorStatisticsDataBuilder();
 435
 436             statisticsDataBuilder.setFlowCapableNodeConnectorStatistics(statisticsBuilder.build());
 437
 438             InstanceIdentifier<NodeConnector> nodeConnectorRef = InstanceIdentifier.builder(Nodes.class)
 439                     .child(Node.class, targetNodeKey)
 440                     .child(NodeConnector.class, new NodeConnectorKey(portStats.getNodeConnectorId())).toInstance();
 441
 442             // FIXME: can we bypass this read?
 443             NodeConnector nodeConnector = (NodeConnector)trans.readOperationalData(nodeConnectorRef);
 444             if(nodeConnector != null){
 445                 final FlowCapableNodeConnectorStatisticsData stats = statisticsDataBuilder.build();
 446                 logger.debug("Augmenting port statistics {} to port {}",stats,nodeConnectorRef.toString());
 447                 NodeConnectorBuilder nodeConnectorBuilder = new NodeConnectorBuilder();
 448                 nodeConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, stats);
 449                 trans.putOperationalData(nodeConnectorRef, nodeConnectorBuilder.build());
 450             }
 451
 452             // FIXME: should we be tracking this data?
 453         }
 454
 455         trans.commit();
 456     }
 457
 458     public synchronized void updateAggregateFlowStats(Short tableId, AggregateFlowStatistics flowStats) {
 459         if (tableId != null) {
 460             final DataModificationTransaction trans = statisticsProvider.startChange();
 461
 462
 463             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 464                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
 465
 466             AggregateFlowStatisticsDataBuilder aggregateFlowStatisticsDataBuilder = new AggregateFlowStatisticsDataBuilder();
 467             AggregateFlowStatisticsBuilder aggregateFlowStatisticsBuilder = new AggregateFlowStatisticsBuilder(flowStats);
 468
 469             aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
 470
 471             logger.debug("Augment aggregate statistics: {} for table {} on Node {}",
 472                     aggregateFlowStatisticsBuilder.build().toString(),tableId,targetNodeKey);
 473
 474             TableBuilder tableBuilder = new TableBuilder();
 475             tableBuilder.setKey(new TableKey(tableId));
 476             tableBuilder.addAugmentation(AggregateFlowStatisticsData.class, aggregateFlowStatisticsDataBuilder.build());
 477             trans.putOperationalData(tableRef, tableBuilder.build());
 478
 479             // FIXME: should we be tracking this data?
 480             trans.commit();
 481         }
 482     }
 483
 484     public synchronized void updateGroupFeatures(GroupFeatures notification) {
 485         final DataModificationTransaction trans = statisticsProvider.startChange();
 486
 487         final NodeBuilder nodeData = new NodeBuilder();
 488         nodeData.setKey(targetNodeKey);
 489
 490         NodeGroupFeaturesBuilder nodeGroupFeatures = new NodeGroupFeaturesBuilder();
 491         GroupFeaturesBuilder groupFeatures = new GroupFeaturesBuilder(notification);
 492         nodeGroupFeatures.setGroupFeatures(groupFeatures.build());
 493
 494         //Update augmented data
 495         nodeData.addAugmentation(NodeGroupFeatures.class, nodeGroupFeatures.build());
 496         trans.putOperationalData(targetNodeIdentifier, nodeData.build());
 497
 498         // FIXME: should we be tracking this data?
 499         trans.commit();
 500     }
 501
 502     public synchronized void updateMeterFeatures(MeterFeatures features) {
 503         final DataModificationTransaction trans = statisticsProvider.startChange();
 504
 505         final NodeBuilder nodeData = new NodeBuilder();
 506         nodeData.setKey(targetNodeKey);
 507
 508         NodeMeterFeaturesBuilder nodeMeterFeatures = new NodeMeterFeaturesBuilder();
 509         MeterFeaturesBuilder meterFeature = new MeterFeaturesBuilder(features);
 510         nodeMeterFeatures.setMeterFeatures(meterFeature.build());
 511
 512         //Update augmented data
 513         nodeData.addAugmentation(NodeMeterFeatures.class, nodeMeterFeatures.build());
 514         trans.putOperationalData(targetNodeIdentifier, nodeData.build());
 515
 516         // FIXME: should we be tracking this data?
 517         trans.commit();
 518     }
 519
 520     public synchronized void updateFlowStats(List<FlowAndStatisticsMapList> list) {
 521         final Long expiryTime = getExpiryTime();
 522         final DataModificationTransaction trans = statisticsProvider.startChange();
 523
 524         for(FlowAndStatisticsMapList map : list) {
 525             short tableId = map.getTableId();
 526             boolean foundOriginalFlow = false;
 527
 528             FlowBuilder flowBuilder = new FlowBuilder();
 529
 530             FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
 531
 532             FlowBuilder flow = new FlowBuilder();
 533             flow.setContainerName(map.getContainerName());
 534             flow.setBufferId(map.getBufferId());
 535             flow.setCookie(map.getCookie());
 536             flow.setCookieMask(map.getCookieMask());
 537             flow.setFlags(map.getFlags());
 538             flow.setFlowName(map.getFlowName());
 539             flow.setHardTimeout(map.getHardTimeout());
 540             if(map.getFlowId() != null)
 541                 flow.setId(new FlowId(map.getFlowId().getValue()));
 542             flow.setIdleTimeout(map.getIdleTimeout());
 543             flow.setInstallHw(map.isInstallHw());
 544             flow.setInstructions(map.getInstructions());
 545             if(map.getFlowId()!= null)
 546                 flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
 547             flow.setMatch(map.getMatch());
 548             flow.setOutGroup(map.getOutGroup());
 549             flow.setOutPort(map.getOutPort());
 550             flow.setPriority(map.getPriority());
 551             flow.setStrict(map.isStrict());
 552             flow.setTableId(tableId);
 553
 554             Flow flowRule = flow.build();
 555
 556             FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
 557             stats.setByteCount(map.getByteCount());
 558             stats.setPacketCount(map.getPacketCount());
 559             stats.setDuration(map.getDuration());
 560
 561             GenericStatistics flowStats = stats.build();
 562
 563             //Augment the data to the flow node
 564
 565             FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
 566             flowStatistics.setByteCount(flowStats.getByteCount());
 567             flowStatistics.setPacketCount(flowStats.getPacketCount());
 568             flowStatistics.setDuration(flowStats.getDuration());
 569             flowStatistics.setContainerName(map.getContainerName());
 570             flowStatistics.setBufferId(map.getBufferId());
 571             flowStatistics.setCookie(map.getCookie());
 572             flowStatistics.setCookieMask(map.getCookieMask());
 573             flowStatistics.setFlags(map.getFlags());
 574             flowStatistics.setFlowName(map.getFlowName());
 575             flowStatistics.setHardTimeout(map.getHardTimeout());
 576             flowStatistics.setIdleTimeout(map.getIdleTimeout());
 577             flowStatistics.setInstallHw(map.isInstallHw());
 578             flowStatistics.setInstructions(map.getInstructions());
 579             flowStatistics.setMatch(map.getMatch());
 580             flowStatistics.setOutGroup(map.getOutGroup());
 581             flowStatistics.setOutPort(map.getOutPort());
 582             flowStatistics.setPriority(map.getPriority());
 583             flowStatistics.setStrict(map.isStrict());
 584             flowStatistics.setTableId(tableId);
 585
 586             flowStatisticsData.setFlowStatistics(flowStatistics.build());
 587
 588             logger.debug("Flow : {}",flowRule.toString());
 589             logger.debug("Statistics to augment : {}",flowStatistics.build().toString());
 590
 591             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 592                     .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
 593
 594             Table table= (Table)trans.readConfigurationData(tableRef);
 595
 596             //TODO: Not a good way to do it, need to figure out better way.
 597             //TODO: major issue in any alternate approach is that flow key is incrementally assigned
 598             //to the flows stored in data store.
 599             // Augment same statistics to all the matching masked flow
 600             if(table != null){
 601
 602                 for(Flow existingFlow : table.getFlow()){
 603                     logger.debug("Existing flow in data store : {}",existingFlow.toString());
 604                     if(FlowComparator.flowEquals(flowRule,existingFlow)){
 605                         InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 606                                 .augmentation(FlowCapableNode.class)
 607                                 .child(Table.class, new TableKey(tableId))
 608                                 .child(Flow.class,existingFlow.getKey()).toInstance();
 609                         flowBuilder.setKey(existingFlow.getKey());
 610                         flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 611                         logger.debug("Found matching flow in the datastore, augmenting statistics");
 612                         foundOriginalFlow = true;
 613                         // Update entry with timestamp of latest response
 614                         flow.setKey(existingFlow.getKey());
 615                         FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
 616                         flowStatsUpdate.put(flowStatsEntry, expiryTime);
 617
 618                         trans.putOperationalData(flowRef, flowBuilder.build());
 619                     }
 620                 }
 621             }
 622
 623             table = (Table)trans.readOperationalData(tableRef);
 624             if(!foundOriginalFlow && table != null){
 625
 626                 for(Flow existingFlow : table.getFlow()){
 627                     FlowStatisticsData augmentedflowStatisticsData = existingFlow.getAugmentation(FlowStatisticsData.class);
 628                     if(augmentedflowStatisticsData != null){
 629                         FlowBuilder existingOperationalFlow = new FlowBuilder();
 630                         existingOperationalFlow.fieldsFrom(augmentedflowStatisticsData.getFlowStatistics());
 631                         logger.debug("Existing unaccounted flow in operational data store : {}",existingFlow.toString());
 632                         if(FlowComparator.flowEquals(flowRule,existingOperationalFlow.build())){
 633                             InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 634                                     .augmentation(FlowCapableNode.class)
 635                                     .child(Table.class, new TableKey(tableId))
 636                                     .child(Flow.class,existingFlow.getKey()).toInstance();
 637                             flowBuilder.setKey(existingFlow.getKey());
 638                             flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 639                             logger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
 640                             foundOriginalFlow = true;
 641
 642                             // Update entry with timestamp of latest response
 643                             flow.setKey(existingFlow.getKey());
 644                             FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
 645                             flowStatsUpdate.put(flowStatsEntry, expiryTime);
 646                             trans.putOperationalData(flowRef, flowBuilder.build());
 647                             break;
 648                         }
 649                     }
 650                 }
 651             }
 652             if(!foundOriginalFlow){
 653                 String flowKey = "#UF$TABLE*"+Short.toString(tableId)+"*"+Integer.toString(this.unaccountedFlowsCounter);
 654                 this.unaccountedFlowsCounter++;
 655                 FlowKey newFlowKey = new FlowKey(new FlowId(flowKey));
 656                 InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
 657                         .augmentation(FlowCapableNode.class)
 658                         .child(Table.class, new TableKey(tableId))
 659                         .child(Flow.class,newFlowKey).toInstance();
 660                 flowBuilder.setKey(newFlowKey);
 661                 flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
 662                 logger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",
 663                         flowBuilder.build());
 664
 665                 // Update entry with timestamp of latest response
 666                 flow.setKey(newFlowKey);
 667                 FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
 668                 flowStatsUpdate.put(flowStatsEntry, expiryTime);
 669                 trans.putOperationalData(flowRef, flowBuilder.build());
 670             }
 671         }
 672
 673         trans.commit();
 674     }
 675
 676     private static Long getExpiryTime(){
 677         final long now = System.nanoTime();
 678         return now + TimeUnit.MILLISECONDS.toNanos(StatisticsProvider.STATS_THREAD_EXECUTION_TIME * NUMBER_OF_WAIT_CYCLES);
 679     }
 680
 681     public synchronized void cleanStaleStatistics(){
 682         final DataModificationTransaction trans = this.statisticsProvider.startChange();
 683         final long now = System.nanoTime();
 684
 685         //Clean stale statistics related to group
 686         for (Iterator<Entry<GroupDescStats, Long>> it = this.groupDescStatsUpdate.entrySet().iterator();it.hasNext();){
 687             Entry<GroupDescStats, Long> e = it.next();
 688             if (now > e.getValue()) {
 689                 cleanGroupStatsFromDataStore(trans, e.getKey());
 690                 it.remove();
 691             }
 692         }
 693
 694         //Clean stale statistics related to meter
 695         for (Iterator<Entry<MeterConfigStats, Long>> it = this.meterConfigStatsUpdate.entrySet().iterator();it.hasNext();){
 696             Entry<MeterConfigStats, Long> e = it.next();
 697             if (now > e.getValue()) {
 698                 cleanMeterStatsFromDataStore(trans, e.getKey());
 699                 it.remove();
 700             }
 701         }
 702
 703         //Clean stale statistics related to flow
 704         for (Iterator<Entry<FlowEntry, Long>> it = this.flowStatsUpdate.entrySet().iterator();it.hasNext();){
 705             Entry<FlowEntry, Long> e = it.next();
 706             if (now > e.getValue()) {
 707                 cleanFlowStatsFromDataStore(trans, e.getKey());
 708                 it.remove();
 709             }
 710         }
 711
 712         //Clean stale statistics related to queue
 713         for (Iterator<Entry<QueueEntry, Long>> it = this.queuesStatsUpdate.entrySet().iterator();it.hasNext();){
 714             Entry<QueueEntry, Long> e = it.next();
 715             if (now > e.getValue()) {
 716                 cleanQueueStatsFromDataStore(trans, e.getKey());
 717                 it.remove();
 718             }
 719         }
 720
 721         trans.commit();
 722     }
 723
 724     private void cleanQueueStatsFromDataStore(DataModificationTransaction trans, QueueEntry queueEntry) {
 725         InstanceIdentifier<?> queueRef
 726                         = InstanceIdentifier.builder(Nodes.class)
 727                                             .child(Node.class, this.targetNodeKey)
 728                                             .child(NodeConnector.class, new NodeConnectorKey(queueEntry.getNodeConnectorId()))
 729                                             .augmentation(FlowCapableNodeConnector.class)
 730                                             .child(Queue.class, new QueueKey(queueEntry.getQueueId()))
 731                                             .augmentation(FlowCapableNodeConnectorQueueStatisticsData.class).toInstance();
 732         trans.removeOperationalData(queueRef);
 733     }
 734
 735     private void cleanFlowStatsFromDataStore(DataModificationTransaction trans, FlowEntry flowEntry) {
 736         InstanceIdentifier<?> flowRef
 737                         = InstanceIdentifier.builder(Nodes.class).child(Node.class, this.targetNodeKey)
 738                                             .augmentation(FlowCapableNode.class)
 739                                             .child(Table.class, new TableKey(flowEntry.getTableId()))
 740                                             .child(Flow.class,flowEntry.getFlow().getKey())
 741                                             .augmentation(FlowStatisticsData.class).toInstance();
 742         trans.removeOperationalData(flowRef);
 743     }
 744
 745     private void cleanMeterStatsFromDataStore(DataModificationTransaction trans, MeterConfigStats meterConfigStats) {
 746         InstanceIdentifierBuilder<Meter> meterRef
 747                         = InstanceIdentifier.builder(Nodes.class).child(Node.class,this.targetNodeKey)
 748                                             .augmentation(FlowCapableNode.class)
 749                                             .child(Meter.class,new MeterKey(meterConfigStats.getMeterId()));
 750
 751         InstanceIdentifier<?> nodeMeterConfigStatsAugmentation = meterRef.augmentation(NodeMeterConfigStats.class).toInstance();
 752         trans.removeOperationalData(nodeMeterConfigStatsAugmentation);
 753
 754         InstanceIdentifier<?> nodeMeterStatisticsAugmentation = meterRef.augmentation(NodeMeterStatistics.class).toInstance();
 755         trans.removeOperationalData(nodeMeterStatisticsAugmentation);
 756     }
 757
 758     private void cleanGroupStatsFromDataStore(DataModificationTransaction trans, GroupDescStats groupDescStats) {
 759         InstanceIdentifierBuilder<Group> groupRef
 760                         = InstanceIdentifier.builder(Nodes.class).child(Node.class,this.targetNodeKey)
 761                                             .augmentation(FlowCapableNode.class)
 762                                             .child(Group.class,new GroupKey(groupDescStats.getGroupId()));
 763
 764         InstanceIdentifier<?> nodeGroupDescStatsAugmentation = groupRef.augmentation(NodeGroupDescStats.class).toInstance();
 765         trans.removeOperationalData(nodeGroupDescStatsAugmentation);
 766
 767         InstanceIdentifier<?> nodeGroupStatisticsAugmentation = groupRef.augmentation(NodeGroupStatistics.class).toInstance();
 768         trans.removeOperationalData(nodeGroupStatisticsAugmentation);
 769     }
 770
 771     @Override
 772     public void close() {
 773         // FIXME: cleanup any resources we hold (registrations, etc.)
 774         logger.debug("Statistics handler for {} shut down", targetNodeKey.getId());
 775     }
 776 }