2 * Copyright IBM Corporation, 2013. All rights reserved.
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
8 package org.opendaylight.controller.md.statistics.manager;
10 import java.util.Collection;
11 import java.util.Collections;
12 import java.util.HashMap;
13 import java.util.HashSet;
14 import java.util.Iterator;
15 import java.util.List;
17 import java.util.Map.Entry;
19 import java.util.concurrent.TimeUnit;
21 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
22 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsData;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsDataBuilder;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsDataBuilder;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.aggregate.flow.statistics.AggregateFlowStatisticsBuilder;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapListBuilder;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.statistics.FlowStatisticsBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsData;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsDataBuilder;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.statistics.FlowTableStatistics;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.statistics.FlowTableStatisticsBuilder;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueBuilder;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.QueueKey;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.queue.rev130925.QueueId;
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.group.desc.GroupDescBuilder;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.features.GroupFeaturesBuilder;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.statistics.GroupStatisticsBuilder;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupFeatures;
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.NodeConnectorId;
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.NodeMeterConfigStats;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterConfigStatsBuilder;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeaturesBuilder;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatistics;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterStatisticsBuilder;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeaturesBuilder;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterConfigStatsBuilder;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.meter.MeterStatisticsBuilder;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterFeatures;
86 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.AggregateFlowStatistics;
89 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.GenericStatistics;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
91 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
92 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.flow.capable.node.connector.statistics.FlowCapableNodeConnectorStatisticsBuilder;
93 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsData;
95 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.FlowCapableNodeConnectorQueueStatisticsDataBuilder;
96 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.flow.capable.node.connector.queue.statistics.FlowCapableNodeConnectorQueueStatisticsBuilder;
97 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.queue.id.and.statistics.map.QueueIdAndStatisticsMap;
98 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
99 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
100 import org.slf4j.Logger;
101 import org.slf4j.LoggerFactory;
103 import com.google.common.base.Preconditions;
106 * This class handles the lifecycle of per-node statistics. It receives data
107 * from StatisticsListener, stores it in the data store and keeps track of
108 * when the data should be removed.
110 * @author avishnoi@in.ibm.com
112 public final class NodeStatisticsHandler implements AutoCloseable {
113 private static final Logger logger = LoggerFactory.getLogger(NodeStatisticsHandler.class);
114 private static final int NUMBER_OF_WAIT_CYCLES = 2;
116 private final Map<GroupDescStats,Long> groupDescStatsUpdate = new HashMap<>();
117 private final Map<MeterConfigStats,Long> meterConfigStatsUpdate = new HashMap<>();
118 private final Map<FlowEntry,Long> flowStatsUpdate = new HashMap<>();
119 private final Map<QueueEntry,Long> queuesStatsUpdate = new HashMap<>();
120 private final InstanceIdentifier<Node> targetNodeIdentifier;
121 private final DataProviderService dps;
122 private final NodeRef targetNodeRef;
123 private final NodeKey targetNodeKey;
124 private Collection<TableKey> knownTables = Collections.emptySet();
125 private int unaccountedFlowsCounter = 1;
127 public NodeStatisticsHandler(final DataProviderService dps, final NodeKey nodeKey) {
128 this.dps = Preconditions.checkNotNull(dps);
129 this.targetNodeKey = Preconditions.checkNotNull(nodeKey);
130 this.targetNodeIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).build();
131 this.targetNodeRef = new NodeRef(targetNodeIdentifier);
134 private static class FlowEntry {
135 private final Short tableId;
136 private final Flow flow;
138 public FlowEntry(Short tableId, Flow flow){
139 this.tableId = tableId;
143 public Short getTableId() {
147 public Flow getFlow() {
152 public int hashCode() {
153 final int prime = 31;
155 result = prime * result + ((flow == null) ? 0 : flow.hashCode());
156 result = prime * result + ((tableId == null) ? 0 : tableId.hashCode());
161 public boolean equals(Object obj) {
166 if (getClass() != obj.getClass())
168 FlowEntry other = (FlowEntry) obj;
170 if (other.flow != null)
172 } else if (!flow.equals(other.flow))
174 if (tableId == null) {
175 if (other.tableId != null)
177 } else if (!tableId.equals(other.tableId))
183 private static final class QueueEntry {
184 private final NodeConnectorId nodeConnectorId;
185 private final QueueId queueId;
186 public QueueEntry(NodeConnectorId ncId, QueueId queueId){
187 this.nodeConnectorId = ncId;
188 this.queueId = queueId;
190 public NodeConnectorId getNodeConnectorId() {
191 return nodeConnectorId;
193 public QueueId getQueueId() {
197 public int hashCode() {
198 final int prime = 31;
200 result = prime * result + ((nodeConnectorId == null) ? 0 : nodeConnectorId.hashCode());
201 result = prime * result + ((queueId == null) ? 0 : queueId.hashCode());
205 public boolean equals(Object obj) {
212 if (!(obj instanceof QueueEntry)) {
215 QueueEntry other = (QueueEntry) obj;
216 if (nodeConnectorId == null) {
217 if (other.nodeConnectorId != null) {
220 } else if (!nodeConnectorId.equals(other.nodeConnectorId)) {
223 if (queueId == null) {
224 if (other.queueId != null) {
227 } else if (!queueId.equals(other.queueId)) {
234 public NodeKey getTargetNodeKey() {
235 return targetNodeKey;
238 public Collection<TableKey> getKnownTables() {
242 public InstanceIdentifier<Node> getTargetNodeIdentifier() {
243 return targetNodeIdentifier;
246 public NodeRef getTargetNodeRef() {
247 return targetNodeRef;
250 public synchronized void updateGroupDescStats(List<GroupDescStats> list){
251 final Long expiryTime = getExpiryTime();
252 final DataModificationTransaction trans = dps.beginTransaction();
254 for (GroupDescStats groupDescStats : list) {
255 GroupBuilder groupBuilder = new GroupBuilder();
256 GroupKey groupKey = new GroupKey(groupDescStats.getGroupId());
257 groupBuilder.setKey(groupKey);
259 InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
260 .augmentation(FlowCapableNode.class)
261 .child(Group.class,groupKey).toInstance();
263 NodeGroupDescStatsBuilder groupDesc= new NodeGroupDescStatsBuilder();
264 GroupDescBuilder stats = new GroupDescBuilder();
265 stats.fieldsFrom(groupDescStats);
266 groupDesc.setGroupDesc(stats.build());
268 //Update augmented data
269 groupBuilder.addAugmentation(NodeGroupDescStats.class, groupDesc.build());
271 trans.putOperationalData(groupRef, groupBuilder.build());
272 this.groupDescStatsUpdate.put(groupDescStats, expiryTime);
278 public synchronized void updateGroupStats(List<GroupStats> list) {
279 final DataModificationTransaction trans = dps.beginTransaction();
281 for(GroupStats groupStats : list) {
282 GroupBuilder groupBuilder = new GroupBuilder();
283 GroupKey groupKey = new GroupKey(groupStats.getGroupId());
284 groupBuilder.setKey(groupKey);
286 InstanceIdentifier<Group> groupRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
287 .augmentation(FlowCapableNode.class)
288 .child(Group.class,groupKey).toInstance();
290 NodeGroupStatisticsBuilder groupStatisticsBuilder= new NodeGroupStatisticsBuilder();
291 GroupStatisticsBuilder stats = new GroupStatisticsBuilder();
292 stats.fieldsFrom(groupStats);
293 groupStatisticsBuilder.setGroupStatistics(stats.build());
295 //Update augmented data
296 groupBuilder.addAugmentation(NodeGroupStatistics.class, groupStatisticsBuilder.build());
297 trans.putOperationalData(groupRef, groupBuilder.build());
299 // FIXME: should we be tracking this data?
305 public synchronized void updateMeterConfigStats(List<MeterConfigStats> list) {
306 final Long expiryTime = getExpiryTime();
307 final DataModificationTransaction trans = dps.beginTransaction();
309 for(MeterConfigStats meterConfigStats : list) {
310 MeterBuilder meterBuilder = new MeterBuilder();
311 MeterKey meterKey = new MeterKey(meterConfigStats.getMeterId());
312 meterBuilder.setKey(meterKey);
314 InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
315 .augmentation(FlowCapableNode.class)
316 .child(Meter.class,meterKey).toInstance();
318 NodeMeterConfigStatsBuilder meterConfig= new NodeMeterConfigStatsBuilder();
319 MeterConfigStatsBuilder stats = new MeterConfigStatsBuilder();
320 stats.fieldsFrom(meterConfigStats);
321 meterConfig.setMeterConfigStats(stats.build());
323 //Update augmented data
324 meterBuilder.addAugmentation(NodeMeterConfigStats.class, meterConfig.build());
326 trans.putOperationalData(meterRef, meterBuilder.build());
327 this.meterConfigStatsUpdate.put(meterConfigStats, expiryTime);
334 public synchronized void updateMeterStats(List<MeterStats> list) {
335 final DataModificationTransaction trans = dps.beginTransaction();
337 for(MeterStats meterStats : list) {
338 MeterBuilder meterBuilder = new MeterBuilder();
339 MeterKey meterKey = new MeterKey(meterStats.getMeterId());
340 meterBuilder.setKey(meterKey);
342 InstanceIdentifier<Meter> meterRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
343 .augmentation(FlowCapableNode.class)
344 .child(Meter.class,meterKey).toInstance();
346 NodeMeterStatisticsBuilder meterStatsBuilder= new NodeMeterStatisticsBuilder();
347 MeterStatisticsBuilder stats = new MeterStatisticsBuilder();
348 stats.fieldsFrom(meterStats);
349 meterStatsBuilder.setMeterStatistics(stats.build());
351 //Update augmented data
352 meterBuilder.addAugmentation(NodeMeterStatistics.class, meterStatsBuilder.build());
353 trans.putOperationalData(meterRef, meterBuilder.build());
355 // FIXME: should we be tracking this data?
361 public synchronized void updateQueueStats(List<QueueIdAndStatisticsMap> list) {
362 final Long expiryTime = getExpiryTime();
363 final DataModificationTransaction trans = dps.beginTransaction();
365 for (QueueIdAndStatisticsMap swQueueStats : list) {
367 QueueEntry queueEntry = new QueueEntry(swQueueStats.getNodeConnectorId(),swQueueStats.getQueueId());
369 FlowCapableNodeConnectorQueueStatisticsDataBuilder queueStatisticsDataBuilder = new FlowCapableNodeConnectorQueueStatisticsDataBuilder();
371 FlowCapableNodeConnectorQueueStatisticsBuilder queueStatisticsBuilder = new FlowCapableNodeConnectorQueueStatisticsBuilder();
373 queueStatisticsBuilder.fieldsFrom(swQueueStats);
375 queueStatisticsDataBuilder.setFlowCapableNodeConnectorQueueStatistics(queueStatisticsBuilder.build());
377 InstanceIdentifier<Queue> queueRef
378 = InstanceIdentifier.builder(Nodes.class)
379 .child(Node.class, targetNodeKey)
380 .child(NodeConnector.class, new NodeConnectorKey(swQueueStats.getNodeConnectorId()))
381 .augmentation(FlowCapableNodeConnector.class)
382 .child(Queue.class, new QueueKey(swQueueStats.getQueueId())).toInstance();
384 QueueBuilder queueBuilder = new QueueBuilder();
385 FlowCapableNodeConnectorQueueStatisticsData qsd = queueStatisticsDataBuilder.build();
386 queueBuilder.addAugmentation(FlowCapableNodeConnectorQueueStatisticsData.class, qsd);
387 queueBuilder.setKey(new QueueKey(swQueueStats.getQueueId()));
389 logger.debug("Augmenting queue statistics {} of queue {} to port {}",
391 swQueueStats.getQueueId(),
392 swQueueStats.getNodeConnectorId());
394 trans.putOperationalData(queueRef, queueBuilder.build());
395 this.queuesStatsUpdate.put(queueEntry, expiryTime);
401 public synchronized void updateFlowTableStats(List<FlowTableAndStatisticsMap> list) {
402 final DataModificationTransaction trans = dps.beginTransaction();
404 final Set<TableKey> knownTables = new HashSet<>(list.size());
405 for (FlowTableAndStatisticsMap ftStats : list) {
407 InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
408 .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(ftStats.getTableId().getValue())).toInstance();
410 FlowTableStatisticsDataBuilder statisticsDataBuilder = new FlowTableStatisticsDataBuilder();
411 final FlowTableStatistics stats = new FlowTableStatisticsBuilder(ftStats).build();
412 statisticsDataBuilder.setFlowTableStatistics(stats);
414 logger.debug("Augment flow table statistics: {} for table {} on Node {}",
415 stats,ftStats.getTableId(), targetNodeKey);
417 TableBuilder tableBuilder = new TableBuilder();
418 tableBuilder.setKey(new TableKey(ftStats.getTableId().getValue()));
419 tableBuilder.addAugmentation(FlowTableStatisticsData.class, statisticsDataBuilder.build());
420 trans.putOperationalData(tableRef, tableBuilder.build());
422 knownTables.add(tableBuilder.getKey());
425 this.knownTables = Collections.unmodifiableCollection(knownTables);
429 public synchronized void updateNodeConnectorStats(List<NodeConnectorStatisticsAndPortNumberMap> list) {
430 final DataModificationTransaction trans = dps.beginTransaction();
432 for(NodeConnectorStatisticsAndPortNumberMap portStats : list) {
434 FlowCapableNodeConnectorStatisticsBuilder statisticsBuilder
435 = new FlowCapableNodeConnectorStatisticsBuilder();
436 statisticsBuilder.setBytes(portStats.getBytes());
437 statisticsBuilder.setCollisionCount(portStats.getCollisionCount());
438 statisticsBuilder.setDuration(portStats.getDuration());
439 statisticsBuilder.setPackets(portStats.getPackets());
440 statisticsBuilder.setReceiveCrcError(portStats.getReceiveCrcError());
441 statisticsBuilder.setReceiveDrops(portStats.getReceiveDrops());
442 statisticsBuilder.setReceiveErrors(portStats.getReceiveErrors());
443 statisticsBuilder.setReceiveFrameError(portStats.getReceiveFrameError());
444 statisticsBuilder.setReceiveOverRunError(portStats.getReceiveOverRunError());
445 statisticsBuilder.setTransmitDrops(portStats.getTransmitDrops());
446 statisticsBuilder.setTransmitErrors(portStats.getTransmitErrors());
448 //Augment data to the node-connector
449 FlowCapableNodeConnectorStatisticsDataBuilder statisticsDataBuilder =
450 new FlowCapableNodeConnectorStatisticsDataBuilder();
452 statisticsDataBuilder.setFlowCapableNodeConnectorStatistics(statisticsBuilder.build());
454 InstanceIdentifier<NodeConnector> nodeConnectorRef = InstanceIdentifier.builder(Nodes.class)
455 .child(Node.class, targetNodeKey)
456 .child(NodeConnector.class, new NodeConnectorKey(portStats.getNodeConnectorId())).toInstance();
458 // FIXME: can we bypass this read?
459 NodeConnector nodeConnector = (NodeConnector)trans.readOperationalData(nodeConnectorRef);
460 if(nodeConnector != null){
461 final FlowCapableNodeConnectorStatisticsData stats = statisticsDataBuilder.build();
462 logger.debug("Augmenting port statistics {} to port {}",stats,nodeConnectorRef.toString());
463 NodeConnectorBuilder nodeConnectorBuilder = new NodeConnectorBuilder();
464 nodeConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, stats);
465 trans.putOperationalData(nodeConnectorRef, nodeConnectorBuilder.build());
468 // FIXME: should we be tracking this data?
474 public synchronized void updateAggregateFlowStats(Short tableId, AggregateFlowStatistics flowStats) {
475 if (tableId != null) {
476 final DataModificationTransaction trans = dps.beginTransaction();
479 InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
480 .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
482 AggregateFlowStatisticsDataBuilder aggregateFlowStatisticsDataBuilder = new AggregateFlowStatisticsDataBuilder();
483 AggregateFlowStatisticsBuilder aggregateFlowStatisticsBuilder = new AggregateFlowStatisticsBuilder(flowStats);
485 aggregateFlowStatisticsDataBuilder.setAggregateFlowStatistics(aggregateFlowStatisticsBuilder.build());
487 logger.debug("Augment aggregate statistics: {} for table {} on Node {}",
488 aggregateFlowStatisticsBuilder.build().toString(),tableId,targetNodeKey);
490 TableBuilder tableBuilder = new TableBuilder();
491 tableBuilder.setKey(new TableKey(tableId));
492 tableBuilder.addAugmentation(AggregateFlowStatisticsData.class, aggregateFlowStatisticsDataBuilder.build());
493 trans.putOperationalData(tableRef, tableBuilder.build());
495 // FIXME: should we be tracking this data?
500 public synchronized void updateGroupFeatures(GroupFeatures notification) {
501 final DataModificationTransaction trans = dps.beginTransaction();
503 final NodeBuilder nodeData = new NodeBuilder();
504 nodeData.setKey(targetNodeKey);
506 NodeGroupFeaturesBuilder nodeGroupFeatures = new NodeGroupFeaturesBuilder();
507 GroupFeaturesBuilder groupFeatures = new GroupFeaturesBuilder(notification);
508 nodeGroupFeatures.setGroupFeatures(groupFeatures.build());
510 //Update augmented data
511 nodeData.addAugmentation(NodeGroupFeatures.class, nodeGroupFeatures.build());
512 trans.putOperationalData(targetNodeIdentifier, nodeData.build());
514 // FIXME: should we be tracking this data?
518 public synchronized void updateMeterFeatures(MeterFeatures features) {
519 final DataModificationTransaction trans = dps.beginTransaction();
521 final NodeBuilder nodeData = new NodeBuilder();
522 nodeData.setKey(targetNodeKey);
524 NodeMeterFeaturesBuilder nodeMeterFeatures = new NodeMeterFeaturesBuilder();
525 MeterFeaturesBuilder meterFeature = new MeterFeaturesBuilder(features);
526 nodeMeterFeatures.setMeterFeatures(meterFeature.build());
528 //Update augmented data
529 nodeData.addAugmentation(NodeMeterFeatures.class, nodeMeterFeatures.build());
530 trans.putOperationalData(targetNodeIdentifier, nodeData.build());
532 // FIXME: should we be tracking this data?
536 public synchronized void updateFlowStats(List<FlowAndStatisticsMapList> list) {
537 final Long expiryTime = getExpiryTime();
538 final DataModificationTransaction trans = dps.beginTransaction();
540 for(FlowAndStatisticsMapList map : list) {
541 short tableId = map.getTableId();
542 boolean foundOriginalFlow = false;
544 FlowBuilder flowBuilder = new FlowBuilder();
546 FlowStatisticsDataBuilder flowStatisticsData = new FlowStatisticsDataBuilder();
548 FlowBuilder flow = new FlowBuilder();
549 flow.setContainerName(map.getContainerName());
550 flow.setBufferId(map.getBufferId());
551 flow.setCookie(map.getCookie());
552 flow.setCookieMask(map.getCookieMask());
553 flow.setFlags(map.getFlags());
554 flow.setFlowName(map.getFlowName());
555 flow.setHardTimeout(map.getHardTimeout());
556 if(map.getFlowId() != null)
557 flow.setId(new FlowId(map.getFlowId().getValue()));
558 flow.setIdleTimeout(map.getIdleTimeout());
559 flow.setInstallHw(map.isInstallHw());
560 flow.setInstructions(map.getInstructions());
561 if(map.getFlowId()!= null)
562 flow.setKey(new FlowKey(new FlowId(map.getKey().getFlowId().getValue())));
563 flow.setMatch(map.getMatch());
564 flow.setOutGroup(map.getOutGroup());
565 flow.setOutPort(map.getOutPort());
566 flow.setPriority(map.getPriority());
567 flow.setStrict(map.isStrict());
568 flow.setTableId(tableId);
570 Flow flowRule = flow.build();
572 FlowAndStatisticsMapListBuilder stats = new FlowAndStatisticsMapListBuilder();
573 stats.setByteCount(map.getByteCount());
574 stats.setPacketCount(map.getPacketCount());
575 stats.setDuration(map.getDuration());
577 GenericStatistics flowStats = stats.build();
579 //Augment the data to the flow node
581 FlowStatisticsBuilder flowStatistics = new FlowStatisticsBuilder();
582 flowStatistics.setByteCount(flowStats.getByteCount());
583 flowStatistics.setPacketCount(flowStats.getPacketCount());
584 flowStatistics.setDuration(flowStats.getDuration());
585 flowStatistics.setContainerName(map.getContainerName());
586 flowStatistics.setBufferId(map.getBufferId());
587 flowStatistics.setCookie(map.getCookie());
588 flowStatistics.setCookieMask(map.getCookieMask());
589 flowStatistics.setFlags(map.getFlags());
590 flowStatistics.setFlowName(map.getFlowName());
591 flowStatistics.setHardTimeout(map.getHardTimeout());
592 flowStatistics.setIdleTimeout(map.getIdleTimeout());
593 flowStatistics.setInstallHw(map.isInstallHw());
594 flowStatistics.setInstructions(map.getInstructions());
595 flowStatistics.setMatch(map.getMatch());
596 flowStatistics.setOutGroup(map.getOutGroup());
597 flowStatistics.setOutPort(map.getOutPort());
598 flowStatistics.setPriority(map.getPriority());
599 flowStatistics.setStrict(map.isStrict());
600 flowStatistics.setTableId(tableId);
602 flowStatisticsData.setFlowStatistics(flowStatistics.build());
604 logger.debug("Flow : {}",flowRule.toString());
605 logger.debug("Statistics to augment : {}",flowStatistics.build().toString());
607 InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
608 .augmentation(FlowCapableNode.class).child(Table.class, new TableKey(tableId)).toInstance();
610 Table table= (Table)trans.readConfigurationData(tableRef);
612 //TODO: Not a good way to do it, need to figure out better way.
613 //TODO: major issue in any alternate approach is that flow key is incrementally assigned
614 //to the flows stored in data store.
615 // Augment same statistics to all the matching masked flow
618 for(Flow existingFlow : table.getFlow()){
619 logger.debug("Existing flow in data store : {}",existingFlow.toString());
620 if(FlowComparator.flowEquals(flowRule,existingFlow)){
621 InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
622 .augmentation(FlowCapableNode.class)
623 .child(Table.class, new TableKey(tableId))
624 .child(Flow.class,existingFlow.getKey()).toInstance();
625 flowBuilder.setKey(existingFlow.getKey());
626 flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
627 logger.debug("Found matching flow in the datastore, augmenting statistics");
628 foundOriginalFlow = true;
629 // Update entry with timestamp of latest response
630 flow.setKey(existingFlow.getKey());
631 FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
632 flowStatsUpdate.put(flowStatsEntry, expiryTime);
634 trans.putOperationalData(flowRef, flowBuilder.build());
639 table = (Table)trans.readOperationalData(tableRef);
640 if(!foundOriginalFlow && table != null){
642 for(Flow existingFlow : table.getFlow()){
643 FlowStatisticsData augmentedflowStatisticsData = existingFlow.getAugmentation(FlowStatisticsData.class);
644 if(augmentedflowStatisticsData != null){
645 FlowBuilder existingOperationalFlow = new FlowBuilder();
646 existingOperationalFlow.fieldsFrom(augmentedflowStatisticsData.getFlowStatistics());
647 logger.debug("Existing unaccounted flow in operational data store : {}",existingFlow.toString());
648 if(FlowComparator.flowEquals(flowRule,existingOperationalFlow.build())){
649 InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
650 .augmentation(FlowCapableNode.class)
651 .child(Table.class, new TableKey(tableId))
652 .child(Flow.class,existingFlow.getKey()).toInstance();
653 flowBuilder.setKey(existingFlow.getKey());
654 flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
655 logger.debug("Found matching unaccounted flow in the operational datastore, augmenting statistics");
656 foundOriginalFlow = true;
658 // Update entry with timestamp of latest response
659 flow.setKey(existingFlow.getKey());
660 FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
661 flowStatsUpdate.put(flowStatsEntry, expiryTime);
662 trans.putOperationalData(flowRef, flowBuilder.build());
668 if(!foundOriginalFlow){
669 String flowKey = "#UF$TABLE*"+Short.toString(tableId)+"*"+Integer.toString(this.unaccountedFlowsCounter);
670 this.unaccountedFlowsCounter++;
671 FlowKey newFlowKey = new FlowKey(new FlowId(flowKey));
672 InstanceIdentifier<Flow> flowRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
673 .augmentation(FlowCapableNode.class)
674 .child(Table.class, new TableKey(tableId))
675 .child(Flow.class,newFlowKey).toInstance();
676 flowBuilder.setKey(newFlowKey);
677 flowBuilder.addAugmentation(FlowStatisticsData.class, flowStatisticsData.build());
678 logger.debug("Flow {} is not present in config data store, augmenting statistics as an unaccounted flow",
679 flowBuilder.build());
681 // Update entry with timestamp of latest response
682 flow.setKey(newFlowKey);
683 FlowEntry flowStatsEntry = new FlowEntry(tableId,flow.build());
684 flowStatsUpdate.put(flowStatsEntry, expiryTime);
685 trans.putOperationalData(flowRef, flowBuilder.build());
692 private static Long getExpiryTime(){
693 final long now = System.nanoTime();
694 return now + TimeUnit.MILLISECONDS.toNanos(StatisticsProvider.STATS_COLLECTION_MILLIS * NUMBER_OF_WAIT_CYCLES);
697 public synchronized void cleanStaleStatistics(){
698 final DataModificationTransaction trans = dps.beginTransaction();
699 final long now = System.nanoTime();
701 //Clean stale statistics related to group
702 for (Iterator<Entry<GroupDescStats, Long>> it = this.groupDescStatsUpdate.entrySet().iterator();it.hasNext();){
703 Entry<GroupDescStats, Long> e = it.next();
704 if (now > e.getValue()) {
705 cleanGroupStatsFromDataStore(trans, e.getKey());
710 //Clean stale statistics related to meter
711 for (Iterator<Entry<MeterConfigStats, Long>> it = this.meterConfigStatsUpdate.entrySet().iterator();it.hasNext();){
712 Entry<MeterConfigStats, Long> e = it.next();
713 if (now > e.getValue()) {
714 cleanMeterStatsFromDataStore(trans, e.getKey());
719 //Clean stale statistics related to flow
720 for (Iterator<Entry<FlowEntry, Long>> it = this.flowStatsUpdate.entrySet().iterator();it.hasNext();){
721 Entry<FlowEntry, Long> e = it.next();
722 if (now > e.getValue()) {
723 cleanFlowStatsFromDataStore(trans, e.getKey());
728 //Clean stale statistics related to queue
729 for (Iterator<Entry<QueueEntry, Long>> it = this.queuesStatsUpdate.entrySet().iterator();it.hasNext();){
730 Entry<QueueEntry, Long> e = it.next();
731 if (now > e.getValue()) {
732 cleanQueueStatsFromDataStore(trans, e.getKey());
740 private void cleanQueueStatsFromDataStore(DataModificationTransaction trans, QueueEntry queueEntry) {
741 InstanceIdentifier<?> queueRef
742 = InstanceIdentifier.builder(Nodes.class)
743 .child(Node.class, this.targetNodeKey)
744 .child(NodeConnector.class, new NodeConnectorKey(queueEntry.getNodeConnectorId()))
745 .augmentation(FlowCapableNodeConnector.class)
746 .child(Queue.class, new QueueKey(queueEntry.getQueueId()))
747 .augmentation(FlowCapableNodeConnectorQueueStatisticsData.class).toInstance();
748 trans.removeOperationalData(queueRef);
751 private void cleanFlowStatsFromDataStore(DataModificationTransaction trans, FlowEntry flowEntry) {
752 InstanceIdentifier<?> flowRef
753 = InstanceIdentifier.builder(Nodes.class).child(Node.class, this.targetNodeKey)
754 .augmentation(FlowCapableNode.class)
755 .child(Table.class, new TableKey(flowEntry.getTableId()))
756 .child(Flow.class,flowEntry.getFlow().getKey())
757 .augmentation(FlowStatisticsData.class).toInstance();
758 trans.removeOperationalData(flowRef);
761 private void cleanMeterStatsFromDataStore(DataModificationTransaction trans, MeterConfigStats meterConfigStats) {
762 InstanceIdentifierBuilder<Meter> meterRef
763 = InstanceIdentifier.builder(Nodes.class).child(Node.class,this.targetNodeKey)
764 .augmentation(FlowCapableNode.class)
765 .child(Meter.class,new MeterKey(meterConfigStats.getMeterId()));
767 InstanceIdentifier<?> nodeMeterConfigStatsAugmentation = meterRef.augmentation(NodeMeterConfigStats.class).toInstance();
768 trans.removeOperationalData(nodeMeterConfigStatsAugmentation);
770 InstanceIdentifier<?> nodeMeterStatisticsAugmentation = meterRef.augmentation(NodeMeterStatistics.class).toInstance();
771 trans.removeOperationalData(nodeMeterStatisticsAugmentation);
774 private void cleanGroupStatsFromDataStore(DataModificationTransaction trans, GroupDescStats groupDescStats) {
775 InstanceIdentifierBuilder<Group> groupRef
776 = InstanceIdentifier.builder(Nodes.class).child(Node.class,this.targetNodeKey)
777 .augmentation(FlowCapableNode.class)
778 .child(Group.class,new GroupKey(groupDescStats.getGroupId()));
780 InstanceIdentifier<?> nodeGroupDescStatsAugmentation = groupRef.augmentation(NodeGroupDescStats.class).toInstance();
781 trans.removeOperationalData(nodeGroupDescStatsAugmentation);
783 InstanceIdentifier<?> nodeGroupStatisticsAugmentation = groupRef.augmentation(NodeGroupStatistics.class).toInstance();
784 trans.removeOperationalData(nodeGroupStatisticsAugmentation);
788 public void close() {
789 // FIXME: cleanup any resources we hold (registrations, etc.)
790 logger.debug("Statistics handler for {} shut down", targetNodeKey.getId());