2 * Copyright (c) 2013 Cisco Systems, Inc. and others. 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
9 package org.opendaylight.controller.protocol_plugin.openflow.internal;
11 import java.nio.ByteBuffer;
12 import java.util.ArrayList;
13 import java.util.Collections;
14 import java.util.Deque;
15 import java.util.HashMap;
16 import java.util.HashSet;
17 import java.util.List;
19 import java.util.Map.Entry;
21 import java.util.Timer;
22 import java.util.TimerTask;
23 import java.util.concurrent.BlockingQueue;
24 import java.util.concurrent.ConcurrentHashMap;
25 import java.util.concurrent.ConcurrentMap;
26 import java.util.concurrent.LinkedBlockingDeque;
27 import java.util.concurrent.LinkedBlockingQueue;
29 import org.eclipse.osgi.framework.console.CommandInterpreter;
30 import org.eclipse.osgi.framework.console.CommandProvider;
31 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
32 import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
33 import org.opendaylight.controller.protocol_plugin.openflow.IStatisticsListener;
34 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
35 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
36 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6Match;
37 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsReply;
38 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsRequest;
39 import org.opendaylight.controller.sal.core.Node;
40 import org.opendaylight.controller.sal.core.NodeConnector;
41 import org.opendaylight.controller.sal.core.Property;
42 import org.opendaylight.controller.sal.core.UpdateType;
43 import org.opendaylight.controller.sal.utils.HexEncode;
44 import org.openflow.protocol.OFError;
45 import org.openflow.protocol.OFMatch;
46 import org.openflow.protocol.OFPort;
47 import org.openflow.protocol.OFStatisticsRequest;
48 import org.openflow.protocol.statistics.OFAggregateStatisticsRequest;
49 import org.openflow.protocol.statistics.OFDescriptionStatistics;
50 import org.openflow.protocol.statistics.OFFlowStatisticsReply;
51 import org.openflow.protocol.statistics.OFFlowStatisticsRequest;
52 import org.openflow.protocol.statistics.OFPortStatisticsReply;
53 import org.openflow.protocol.statistics.OFPortStatisticsRequest;
54 import org.openflow.protocol.statistics.OFQueueStatisticsRequest;
55 import org.openflow.protocol.statistics.OFStatistics;
56 import org.openflow.protocol.statistics.OFStatisticsType;
57 import org.openflow.protocol.statistics.OFTableStatistics;
58 import org.openflow.protocol.statistics.OFVendorStatistics;
59 import org.openflow.util.HexString;
60 import org.osgi.framework.BundleContext;
61 import org.osgi.framework.FrameworkUtil;
62 import org.slf4j.Logger;
63 import org.slf4j.LoggerFactory;
66 * It periodically polls the different OF statistics from the OF switches and
67 * caches them for quick retrieval for the above layers' modules It also
68 * provides an API to directly query the switch about the statistics
70 public class OFStatisticsManager implements IOFStatisticsManager,
71 IInventoryShimExternalListener, CommandProvider {
72 private static final Logger log = LoggerFactory
73 .getLogger(OFStatisticsManager.class);
74 private static final int initialSize = 64;
75 private static final long flowStatsPeriod = 10000;
76 private static final long descriptionStatsPeriod = 60000;
77 private static final long portStatsPeriod = 5000;
78 private static final long tableStatsPeriod = 10000;
79 private static final long tickPeriod = 1000;
80 private static short statisticsTickNumber = (short) (flowStatsPeriod / tickPeriod);
81 private static short descriptionTickNumber = (short) (descriptionStatsPeriod / tickPeriod);
82 private static short portTickNumber = (short) (portStatsPeriod / tickPeriod);
83 private static short tableTickNumber = (short) (tableStatsPeriod / tickPeriod);
84 private static short factoredSamples = (short) 2;
85 private static short counter = 1;
86 private IController controller = null;
87 private ConcurrentMap<Long, List<OFStatistics>> flowStatistics;
88 private ConcurrentMap<Long, List<OFStatistics>> descStatistics;
89 private ConcurrentMap<Long, List<OFStatistics>> portStatistics;
90 private ConcurrentMap<Long, List<OFStatistics>> tableStatistics;
91 private List<OFStatistics> dummyList;
92 private ConcurrentMap<Long, StatisticsTicks> statisticsTimerTicks;
93 protected BlockingQueue<StatsRequest> pendingStatsRequests;
94 protected BlockingQueue<Long> switchPortStatsUpdated;
95 private Thread statisticsCollector;
96 private Thread txRatesUpdater;
97 private Timer statisticsTimer;
98 private TimerTask statisticsTimerTask;
99 private ConcurrentMap<Long, Boolean> switchSupportsVendorExtStats;
100 private Map<Long, Map<Short, TxRates>> txRates; // Per port sampled (every
101 // portStatsPeriod) transmit
103 private Set<IStatisticsListener> descriptionListeners;
106 * The object containing the latest factoredSamples tx rate samples for a
109 protected class TxRates {
110 Deque<Long> sampledTxBytes; // contains the latest factoredSamples
111 // sampled transmitted bytes
114 sampledTxBytes = new LinkedBlockingDeque<Long>();
117 public void update(Long txBytes) {
119 * Based on how many samples our average works on, we might have to
120 * remove the oldest sample
122 if (sampledTxBytes.size() == factoredSamples) {
123 sampledTxBytes.removeLast();
126 // Add the latest sample to the top of the queue
127 sampledTxBytes.addFirst(txBytes);
131 * Returns the average transmit rate in bps
133 * @return the average transmit rate [bps]
135 public long getAverageTxRate() {
138 * If we cannot provide the value for the time window length set
140 if (sampledTxBytes.size() < factoredSamples) {
143 long increment = (long) (sampledTxBytes.getFirst() - sampledTxBytes
145 long timePeriod = (long) (factoredSamples * portStatsPeriod)
147 average = (8L * increment) / timePeriod;
152 public void setController(IController core) {
153 this.controller = core;
156 public void unsetController(IController core) {
157 if (this.controller == core) {
158 this.controller = null;
163 * Function called by the dependency manager when all the required
164 * dependencies are satisfied
168 flowStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
169 descStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
170 portStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
171 tableStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
172 dummyList = new ArrayList<OFStatistics>(1);
173 statisticsTimerTicks = new ConcurrentHashMap<Long, StatisticsTicks>(
175 pendingStatsRequests = new LinkedBlockingQueue<StatsRequest>(
177 switchPortStatsUpdated = new LinkedBlockingQueue<Long>(initialSize);
178 switchSupportsVendorExtStats = new ConcurrentHashMap<Long, Boolean>(
180 txRates = new HashMap<Long, Map<Short, TxRates>>(initialSize);
181 descriptionListeners = new HashSet<IStatisticsListener>();
183 configStatsPollIntervals();
185 // Initialize managed timers
186 statisticsTimer = new Timer();
187 statisticsTimerTask = new TimerTask() {
194 // Initialize Statistics collector thread
195 statisticsCollector = new Thread(new Runnable() {
200 StatsRequest req = pendingStatsRequests.take();
201 acquireStatistics(req.switchId, req.type);
202 } catch (InterruptedException e) {
203 log.warn("Flow Statistics Collector thread "
208 }, "Statistics Collector");
210 // Initialize Tx Rate Updater thread
211 txRatesUpdater = new Thread(new Runnable() {
216 long switchId = switchPortStatsUpdated.take();
217 updatePortsTxRate(switchId);
218 } catch (InterruptedException e) {
219 log.warn("TX Rate Updater thread interrupted", e);
223 }, "TX Rate Updater");
227 * Function called by the dependency manager when at least one dependency
228 * become unsatisfied or when the component is shutting down because for
229 * example bundle is being stopped.
236 * Function called by dependency manager after "init ()" is called and after
237 * the services provided by the class are registered in the service registry
241 // Start managed timers
242 statisticsTimer.scheduleAtFixedRate(statisticsTimerTask, 0, tickPeriod);
244 // Start statistics collector thread
245 statisticsCollector.start();
247 // Start bandwidth utilization computer thread
248 txRatesUpdater.start();
251 registerWithOSGIConsole();
255 * Function called by the dependency manager before the services exported by
256 * the component are unregistered, this will be followed by a "destroy ()"
261 // Stop managed timers
262 statisticsTimer.cancel();
265 public void setStatisticsListener(IStatisticsListener s) {
266 this.descriptionListeners.add(s);
269 public void unsetStatisticsListener(IStatisticsListener s) {
271 this.descriptionListeners.remove(s);
275 private void registerWithOSGIConsole() {
276 BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
278 bundleContext.registerService(CommandProvider.class.getName(), this,
282 private static class StatsRequest {
283 protected Long switchId;
284 protected OFStatisticsType type;
286 public StatsRequest(Long d, OFStatisticsType t) {
292 public String toString() {
293 return "SReq = {switchId=" + switchId + ", type=" + type + "}";
297 public int hashCode() {
298 final int prime = 31;
300 result = prime * result
301 + ((switchId == null) ? 0 : switchId.hashCode());
302 result = prime * result + ((type == null) ? 0 : type.ordinal());
307 public boolean equals(Object obj) {
314 if (getClass() != obj.getClass()) {
317 StatsRequest other = (StatsRequest) obj;
318 if (switchId == null) {
319 if (other.switchId != null) {
322 } else if (!switchId.equals(other.switchId)) {
325 if (type != other.type) {
332 private void addStatisticsTicks(Long switchId) {
333 switchSupportsVendorExtStats.put(switchId, Boolean.TRUE); // Assume
339 statisticsTimerTicks.put(switchId, new StatisticsTicks(true));
340 log.info("Added Switch {} to target pool",
341 HexString.toHexString(switchId.longValue()));
344 protected static class StatisticsTicks {
345 private short flowStatisticsTicks;
346 private short descriptionTicks;
347 private short portStatisticsTicks;
348 private short tableStatisticsTicks;
350 public StatisticsTicks(boolean scattered) {
352 // scatter bursts by statisticsTickPeriod
355 } // being paranoid here
356 flowStatisticsTicks = (short) (1 + counter
357 % statisticsTickNumber);
358 descriptionTicks = (short) (1 + counter % descriptionTickNumber);
359 portStatisticsTicks = (short) (1 + counter % portTickNumber);
360 tableStatisticsTicks = (short) (1 + counter % tableTickNumber);
362 flowStatisticsTicks = statisticsTickNumber;
363 descriptionTicks = descriptionTickNumber;
364 portStatisticsTicks = portTickNumber;
365 tableStatisticsTicks = tableTickNumber;
369 public boolean decrementFlowTicksIsZero() {
370 // Please ensure no code is inserted between the if check and the
371 // flowStatisticsTicks reset
372 if (--flowStatisticsTicks == 0) {
373 flowStatisticsTicks = statisticsTickNumber;
379 public boolean decrementDescTicksIsZero() {
380 // Please ensure no code is inserted between the if check and the
381 // descriptionTicks reset
382 if (--descriptionTicks == 0) {
383 descriptionTicks = descriptionTickNumber;
389 public boolean decrementPortTicksIsZero() {
390 // Please ensure no code is inserted between the if check and the
391 // descriptionTicks reset
392 if (--portStatisticsTicks == 0) {
393 portStatisticsTicks = portTickNumber;
399 public boolean decrementTableTicksIsZero() {
400 // Please ensure no code is inserted between the if check and the
401 // descriptionTicks reset
402 if(--tableStatisticsTicks == 0) {
403 tableStatisticsTicks = tableTickNumber;
410 public String toString() {
411 return "{fT=" + flowStatisticsTicks + ",dT=" + descriptionTicks
412 + ",pT=" + portStatisticsTicks + ",tT=" + tableStatisticsTicks + "}";
416 private void printInfoMessage(String type, StatsRequest request) {
418 "{} stats request not inserted for switch: {}. Queue size: {}. Collector state: {}.",
419 new Object[] { type, HexString.toHexString(request.switchId),
420 pendingStatsRequests.size(),
421 statisticsCollector.getState().toString() });
424 protected void decrementTicks() {
425 StatsRequest request = null;
426 for (Map.Entry<Long, StatisticsTicks> entry : statisticsTimerTicks
428 StatisticsTicks clock = entry.getValue();
429 Long switchId = entry.getKey();
430 if (clock.decrementFlowTicksIsZero() == true) {
431 request = (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) ? new StatsRequest(
432 switchId, OFStatisticsType.VENDOR) : new StatsRequest(
433 switchId, OFStatisticsType.FLOW);
434 // If a request for this switch is already in the queue, skip to
435 // add this new request
436 if (!pendingStatsRequests.contains(request)
437 && false == pendingStatsRequests.offer(request)) {
438 printInfoMessage("Flow", request);
442 if (clock.decrementDescTicksIsZero() == true) {
443 request = new StatsRequest(switchId, OFStatisticsType.DESC);
444 // If a request for this switch is already in the queue, skip to
445 // add this new request
446 if (!pendingStatsRequests.contains(request)
447 && false == pendingStatsRequests.offer(request)) {
448 printInfoMessage("Description", request);
452 if (clock.decrementPortTicksIsZero() == true) {
453 request = new StatsRequest(switchId, OFStatisticsType.PORT);
454 // If a request for this switch is already in the queue, skip to
455 // add this new request
456 if (!pendingStatsRequests.contains(request)
457 && false == pendingStatsRequests.offer(request)) {
458 printInfoMessage("Port", request);
462 if(clock.decrementTableTicksIsZero() == true) {
463 request = new StatsRequest(switchId, OFStatisticsType.TABLE);
464 // If a request for this switch is already in the queue, skip to
465 // add this new request
466 if (!pendingStatsRequests.contains(request)
467 && false == pendingStatsRequests.offer(request)) {
468 printInfoMessage("Table", request);
474 private void removeStatsRequestTasks(Long switchId) {
475 log.info("Cleaning Statistics database for switch {}",
476 HexEncode.longToHexString(switchId));
477 // To be safe, let's attempt removal of both VENDOR and FLOW request. It
479 pendingStatsRequests.remove(new StatsRequest(switchId,
480 OFStatisticsType.VENDOR));
481 pendingStatsRequests.remove(new StatsRequest(switchId,
482 OFStatisticsType.FLOW));
483 pendingStatsRequests.remove(new StatsRequest(switchId,
484 OFStatisticsType.DESC));
485 pendingStatsRequests.remove(new StatsRequest(switchId,
486 OFStatisticsType.PORT));
487 pendingStatsRequests.remove(new StatsRequest(switchId,
488 OFStatisticsType.TABLE));
489 // Take care of the TX rate databases
490 switchPortStatsUpdated.remove(switchId);
491 txRates.remove(switchId);
494 private void clearFlowStatsAndTicks(Long switchId) {
495 statisticsTimerTicks.remove(switchId);
496 removeStatsRequestTasks(switchId);
497 flowStatistics.remove(switchId);
498 log.info("Statistics removed for switch {}",
499 HexString.toHexString(switchId));
502 private void acquireStatistics(Long switchId, OFStatisticsType statType) {
504 // Query the switch on all matches
505 List<OFStatistics> values = this.acquireStatistics(switchId, statType,
508 // Update local caching database if got a valid response
509 if (values != null && !values.isEmpty()) {
510 if ((statType == OFStatisticsType.FLOW)
511 || (statType == OFStatisticsType.VENDOR)) {
512 flowStatistics.put(switchId, values);
513 } else if (statType == OFStatisticsType.DESC) {
514 // Notify who may be interested in a description change
515 notifyDescriptionListeners(switchId, values);
518 descStatistics.put(switchId, values);
519 } else if (statType == OFStatisticsType.PORT) {
520 // Overwrite cache with new port statistics for this switch
521 portStatistics.put(switchId, values);
523 // Wake up the thread which maintains the TX byte counters for
525 switchPortStatsUpdated.offer(switchId);
526 } else if (statType == OFStatisticsType.TABLE) {
528 tableStatistics.put(switchId, values);
533 private void notifyDescriptionListeners(Long switchId,
534 List<OFStatistics> values) {
535 for (IStatisticsListener l : this.descriptionListeners) {
536 l.descriptionRefreshed(switchId,
537 ((OFDescriptionStatistics) values.get(0)));
542 * Generic function to get the statistics form a OF switch
544 @SuppressWarnings("unchecked")
545 private List<OFStatistics> acquireStatistics(Long switchId,
546 OFStatisticsType statsType, Object target) {
547 List<OFStatistics> values = null;
549 ISwitch sw = controller.getSwitch(switchId);
552 OFStatisticsRequest req = new OFStatisticsRequest();
553 req.setStatisticType(statsType);
554 int requestLength = req.getLengthU();
556 if (statsType == OFStatisticsType.FLOW) {
557 OFMatch match = null;
558 if (target == null) {
560 match = new OFMatch();
561 match.setWildcards(0xffffffff);
562 } else if (!(target instanceof OFMatch)) {
564 log.warn("Invalid target type for Flow stats request: {}",
568 // Specific flow request
569 match = (OFMatch) target;
571 OFFlowStatisticsRequest specificReq = new OFFlowStatisticsRequest();
572 specificReq.setMatch(match);
573 specificReq.setOutPort(OFPort.OFPP_NONE.getValue());
574 specificReq.setTableId((byte) 0xff);
575 req.setStatistics(Collections
576 .singletonList((OFStatistics) specificReq));
577 requestLength += specificReq.getLength();
579 } else if (statsType == OFStatisticsType.VENDOR) {
580 V6StatsRequest specificReq = new V6StatsRequest();
581 specificReq.setOutPort(OFPort.OFPP_NONE.getValue());
582 specificReq.setTableId((byte) 0xff);
583 req.setStatistics(Collections
584 .singletonList((OFStatistics) specificReq));
585 requestLength += specificReq.getLength();
587 } else if (statsType == OFStatisticsType.AGGREGATE) {
588 OFAggregateStatisticsRequest specificReq = new OFAggregateStatisticsRequest();
589 OFMatch match = new OFMatch();
590 match.setWildcards(0xffffffff);
591 specificReq.setMatch(match);
592 specificReq.setOutPort(OFPort.OFPP_NONE.getValue());
593 specificReq.setTableId((byte) 0xff);
594 req.setStatistics(Collections
595 .singletonList((OFStatistics) specificReq));
596 requestLength += specificReq.getLength();
598 } else if (statsType == OFStatisticsType.PORT) {
600 if (target == null) {
602 targetPort = (short) OFPort.OFPP_NONE.getValue();
603 } else if (!(target instanceof Short)) {
605 log.warn("Invalid target type for Port stats request: {}",
609 // Specific port request
610 targetPort = (Short) target;
612 OFPortStatisticsRequest specificReq = new OFPortStatisticsRequest();
613 specificReq.setPortNumber(targetPort);
614 req.setStatistics(Collections
615 .singletonList((OFStatistics) specificReq));
616 requestLength += specificReq.getLength();
618 } else if (statsType == OFStatisticsType.QUEUE) {
619 OFQueueStatisticsRequest specificReq = new OFQueueStatisticsRequest();
620 specificReq.setPortNumber((short) OFPort.OFPP_ALL.getValue());
621 specificReq.setQueueId(0xffffffff);
622 req.setStatistics(Collections
623 .singletonList((OFStatistics) specificReq));
624 requestLength += specificReq.getLength();
626 } else if (statsType == OFStatisticsType.DESC) {
628 } else if (statsType == OFStatisticsType.TABLE) {
630 if (!(target instanceof Byte)) {
632 log.warn("Invalid table id for table stats request: {}",
636 byte targetTable = (Byte) target;
637 OFTableStatistics specificReq = new OFTableStatistics();
638 specificReq.setTableId(targetTable);
639 req.setStatistics(Collections
640 .singletonList((OFStatistics) specificReq));
641 requestLength += specificReq.getLength();
645 req.setLengthU(requestLength);
646 Object result = sw.getStatistics(req);
648 if (result == null) {
649 log.warn("Request Timed Out for ({}) from switch {}", type,
650 HexString.toHexString(switchId));
651 } else if (result instanceof OFError) {
652 log.warn("Switch {} failed to handle ({}) stats request: {}",
653 new Object[] { HexString.toHexString(switchId), type,
654 Utils.getOFErrorString((OFError) result) });
655 if (this.switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) {
657 "Switching back to regular Flow stats requests for switch {}",
658 HexString.toHexString(switchId));
659 this.switchSupportsVendorExtStats.put(switchId,
663 values = (List<OFStatistics>) result;
670 public List<OFStatistics> getOFFlowStatistics(Long switchId) {
671 List<OFStatistics> list = flowStatistics.get(switchId);
674 * Check on emptiness as interference between add and get is still
675 * possible on the inner list (the concurrentMap entry's value)
677 return (list == null || list.isEmpty()) ? this.dummyList
678 : (list.get(0) instanceof OFVendorStatistics) ? this
679 .v6StatsListToOFStatsList(list) : list;
683 public List<OFStatistics> getOFFlowStatistics(Long switchId, OFMatch ofMatch) {
684 List<OFStatistics> statsList = flowStatistics.get(switchId);
687 * Check on emptiness as interference between add and get is still
688 * possible on the inner list (the concurrentMap entry's value)
690 if (statsList == null || statsList.isEmpty()) {
691 return this.dummyList;
694 if (statsList.get(0) instanceof OFVendorStatistics) {
696 * Caller could provide regular OF match when we instead pull the
697 * vendor statistics from this node Caller is not supposed to know
698 * whether this switch supports vendor extensions statistics
701 V6Match targetMatch = (ofMatch instanceof V6Match) ? (V6Match) ofMatch
702 : new V6Match(ofMatch);
704 List<OFStatistics> targetList = v6StatsListToOFStatsList(statsList);
705 for (OFStatistics stats : targetList) {
706 V6StatsReply v6Stats = (V6StatsReply) stats;
707 V6Match v6Match = v6Stats.getMatch();
708 if (v6Match.equals(targetMatch)) {
709 List<OFStatistics> list = new ArrayList<OFStatistics>();
715 for (OFStatistics stats : statsList) {
716 OFFlowStatisticsReply flowStats = (OFFlowStatisticsReply) stats;
717 if (flowStats.getMatch().equals(ofMatch)) {
718 List<OFStatistics> list = new ArrayList<OFStatistics>();
724 return this.dummyList;
728 * Converts the v6 vendor statistics to the OFStatistics
730 private List<OFStatistics> v6StatsListToOFStatsList(
731 List<OFStatistics> statistics) {
732 List<OFStatistics> v6statistics = new ArrayList<OFStatistics>();
733 if (statistics != null && !statistics.isEmpty()) {
734 for (OFStatistics stats : statistics) {
735 if (stats instanceof OFVendorStatistics) {
736 List<OFStatistics> r = getV6ReplyStatistics((OFVendorStatistics) stats);
738 v6statistics.addAll(r);
746 private static List<OFStatistics> getV6ReplyStatistics(
747 OFVendorStatistics stat) {
748 int length = stat.getLength();
749 List<OFStatistics> results = new ArrayList<OFStatistics>();
751 return null; // Nicira Hdr is 12 bytes. We need atleast that much
752 ByteBuffer data = ByteBuffer.allocate(length);
755 if (log.isTraceEnabled()) {
756 log.trace("getV6ReplyStatistics: Buffer BYTES ARE {}",
757 HexString.toHexString(data.array()));
760 int vendor = data.getInt(); // first 4 bytes is vendor id.
761 if (vendor != V6StatsRequest.NICIRA_VENDOR_ID) {
762 log.warn("Unexpected vendor id: 0x{}", Integer.toHexString(vendor));
765 // go ahead by 8 bytes which is 8 bytes of 0
766 data.getLong(); // should be all 0's
767 length -= 12; // 4 bytes Nicira Hdr + 8 bytes from above line have
771 V6StatsReply v6statsreply;
774 v6statsreply = new V6StatsReply();
775 min_len = v6statsreply.getLength();
776 if (length < v6statsreply.getLength())
778 v6statsreply.setActionFactory(stat.getActionFactory());
779 v6statsreply.readFrom(data);
780 if (v6statsreply.getLength() < min_len)
782 v6statsreply.setVendorId(vendor);
783 log.trace("V6StatsReply: {}", v6statsreply);
784 length -= v6statsreply.getLength();
785 results.add(v6statsreply);
791 public List<OFStatistics> queryStatistics(Long switchId,
792 OFStatisticsType statType, Object target) {
794 * Caller does not know and it is not supposed to know whether this
795 * switch supports vendor extension. We adjust the target for him
797 if (statType == OFStatisticsType.FLOW) {
798 if (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) {
799 statType = OFStatisticsType.VENDOR;
803 List<OFStatistics> list = this.acquireStatistics(switchId, statType,
806 return (list == null) ? null
807 : (statType == OFStatisticsType.VENDOR) ? v6StatsListToOFStatsList(list)
812 public List<OFStatistics> getOFDescStatistics(Long switchId) {
813 if (!descStatistics.containsKey(switchId))
814 return this.dummyList;
816 return descStatistics.get(switchId);
820 public List<OFStatistics> getOFPortStatistics(Long switchId) {
821 if (!portStatistics.containsKey(switchId)) {
822 return this.dummyList;
825 return portStatistics.get(switchId);
829 public List<OFStatistics> getOFPortStatistics(Long switchId, short portId) {
830 if (!portStatistics.containsKey(switchId)) {
831 return this.dummyList;
833 List<OFStatistics> list = new ArrayList<OFStatistics>(1);
834 for (OFStatistics stats : portStatistics.get(switchId)) {
835 if (((OFPortStatisticsReply) stats).getPortNumber() == portId) {
844 public List<OFStatistics> getOFTableStatistics(Long switchId) {
845 if (!tableStatistics.containsKey(switchId)) {
846 return this.dummyList;
849 return tableStatistics.get(switchId);
853 public List<OFStatistics> getOFTableStatistics(Long switchId, Byte tableId) {
854 if (!tableStatistics.containsKey(switchId)) {
855 return this.dummyList;
858 List<OFStatistics> list = new ArrayList<OFStatistics>(1);
859 for (OFStatistics stats : tableStatistics.get(switchId)) {
860 if (((OFTableStatistics) stats).getTableId() == tableId) {
869 public int getFlowsNumber(long switchId) {
870 return this.flowStatistics.get(switchId).size();
874 * InventoryShim replay for us all the switch addition which happened before
878 public void updateNode(Node node, UpdateType type, Set<Property> props) {
879 Long switchId = (Long) node.getID();
882 addStatisticsTicks(switchId);
885 clearFlowStatsAndTicks(switchId);
891 public void updateNodeConnector(NodeConnector nodeConnector,
892 UpdateType type, Set<Property> props) {
897 * Update the cached port rates for this switch with the latest retrieved
898 * port transmit byte count
902 private synchronized void updatePortsTxRate(long switchId) {
903 List<OFStatistics> newPortStatistics = this.portStatistics
905 if (newPortStatistics == null) {
908 Map<Short, TxRates> rates = this.txRates.get(switchId);
910 // First time rates for this switch are added
911 rates = new HashMap<Short, TxRates>();
912 txRates.put(switchId, rates);
914 for (OFStatistics stats : newPortStatistics) {
915 OFPortStatisticsReply newPortStat = (OFPortStatisticsReply) stats;
916 short port = newPortStat.getPortNumber();
917 TxRates portRatesHolder = rates.get(port);
918 if (portRatesHolder == null) {
919 // First time rates for this port are added
920 portRatesHolder = new TxRates();
921 rates.put(port, portRatesHolder);
923 // Get and store the number of transmitted bytes for this port
924 // And handle the case where agent does not support the counter
925 long transmitBytes = newPortStat.getTransmitBytes();
926 long value = (transmitBytes < 0) ? 0 : transmitBytes;
927 portRatesHolder.update(value);
932 public synchronized long getTransmitRate(Long switchId, Short port) {
934 if (switchId == null || port == null) {
937 Map<Short, TxRates> perSwitch = txRates.get(switchId);
938 if (perSwitch == null) {
941 TxRates portRates = perSwitch.get(port);
942 if (portRates == null) {
945 return portRates.getAverageTxRate();
949 * Manual switch name configuration code
952 public String getHelp() {
953 StringBuffer help = new StringBuffer();
954 help.append("---OF Statistics Manager utilities---\n");
955 help.append("\t ofdumpstatsmgr - "
956 + "Print Internal Stats Mgr db\n");
957 help.append("\t ofstatsmgrintervals <fP> <pP> <dP>(in seconds) - "
958 + "Set/Show flow/port/dedscription stats poll intervals\n");
959 return help.toString();
962 private boolean isValidSwitchId(String switchId) {
963 String regexDatapathID = "^([0-9a-fA-F]{1,2}[:-]){7}[0-9a-fA-F]{1,2}$";
964 String regexDatapathIDLong = "^[0-9a-fA-F]{1,16}$";
966 return (switchId != null && (switchId.matches(regexDatapathID) || switchId
967 .matches(regexDatapathIDLong)));
970 public long getSwitchIDLong(String switchId) {
972 String switchString = "0";
974 if (isValidSwitchId(switchId)) {
975 if (switchId.contains(":")) {
976 // Handle the 00:00:AA:BB:CC:DD:EE:FF notation
977 switchString = switchId.replace(":", "");
978 } else if (switchId.contains("-")) {
979 // Handle the 00-00-AA-BB-CC-DD-EE-FF notation
980 switchString = switchId.replace("-", "");
982 // Handle the 0123456789ABCDEF notation
983 switchString = switchId;
986 return Long.parseLong(switchString, radix);
990 * Internal information dump code
992 private String prettyPrintSwitchMap(ConcurrentMap<Long, StatisticsTicks> map) {
993 StringBuffer buffer = new StringBuffer();
995 for (Entry<Long, StatisticsTicks> entry : map.entrySet()) {
996 buffer.append(HexString.toHexString(entry.getKey()) + "="
997 + entry.getValue().toString() + " ");
1000 return buffer.toString();
1003 public void _ofdumpstatsmgr(CommandInterpreter ci) {
1004 ci.println("Global Counter: " + counter);
1005 ci.println("Timer Ticks: " + prettyPrintSwitchMap(statisticsTimerTicks));
1006 ci.println("PendingStatsQueue: " + pendingStatsRequests);
1007 ci.println("PendingStatsQueue size: " + pendingStatsRequests.size());
1008 ci.println("Stats Collector alive: " + statisticsCollector.isAlive());
1009 ci.println("Stats Collector State: "
1010 + statisticsCollector.getState().toString());
1011 ci.println("StatsTimer: " + statisticsTimer.toString());
1012 ci.println("Flow Stats Period: " + statisticsTickNumber + " s");
1013 ci.println("Desc Stats Period: " + descriptionTickNumber + " s");
1014 ci.println("Port Stats Period: " + portTickNumber + " s");
1015 ci.println("Table Stats Period: " + tableTickNumber + " s");
1018 public void _resetSwitchCapability(CommandInterpreter ci) {
1019 String sidString = ci.nextArgument();
1021 if (sidString == null) {
1022 ci.println("Insert the switch id (numeric value)");
1026 sid = Long.valueOf(sidString);
1027 this.switchSupportsVendorExtStats.put(sid, Boolean.TRUE);
1028 ci.println("Vendor capability for switch " + sid + " set to "
1029 + this.switchSupportsVendorExtStats.get(sid));
1030 } catch (NumberFormatException e) {
1031 ci.println("Invalid switch id. Has to be numeric.");
1036 public void _ofbw(CommandInterpreter ci) {
1037 String sidString = ci.nextArgument();
1039 if (sidString == null) {
1040 ci.println("Insert the switch id (numeric value)");
1044 sid = Long.valueOf(sidString);
1045 } catch (NumberFormatException e) {
1046 ci.println("Invalid switch id. Has to be numeric.");
1049 Map<Short, TxRates> thisSwitchRates = txRates.get(sid);
1050 ci.println("Bandwidth utilization (" + factoredSamples
1051 * portTickNumber + " sec average) for switch "
1052 + HexEncode.longToHexString(sid) + ":");
1053 if (thisSwitchRates == null) {
1054 ci.println("Not available");
1056 for (Entry<Short, TxRates> entry : thisSwitchRates.entrySet()) {
1057 ci.println("Port: " + entry.getKey() + ": "
1058 + entry.getValue().getAverageTxRate() + " bps");
1064 public void _txratewindow(CommandInterpreter ci) {
1065 String averageWindow = ci.nextArgument();
1067 if (averageWindow == null) {
1068 ci.println("Insert the length in seconds of the median "
1069 + "window for tx rate");
1070 ci.println("Current: " + factoredSamples * portTickNumber + " secs");
1074 seconds = Short.valueOf(averageWindow);
1075 } catch (NumberFormatException e) {
1076 ci.println("Invalid period.");
1078 OFStatisticsManager.factoredSamples = (short) (seconds / portTickNumber);
1079 ci.println("New: " + factoredSamples * portTickNumber + " secs");
1082 public void _ofstatsmgrintervals(CommandInterpreter ci) {
1083 String flowStatsInterv = ci.nextArgument();
1084 String portStatsInterv = ci.nextArgument();
1085 String descStatsInterv = ci.nextArgument();
1086 String tableStatsInterv = ci.nextArgument();
1088 if (flowStatsInterv == null || portStatsInterv == null
1089 || descStatsInterv == null) {
1090 ci.println("Usage: ostatsmgrintervals <fP> <pP> <dP>(in seconds)");
1091 ci.println("Current Values: fP=" + statisticsTickNumber + "s pP="
1092 + portTickNumber + "s dP=" + descriptionTickNumber + "s");
1095 Short fP, pP, dP, tP;
1097 fP = Short.parseShort(flowStatsInterv);
1098 pP = Short.parseShort(portStatsInterv);
1099 dP = Short.parseShort(descStatsInterv);
1100 tP = Short.parseShort(tableStatsInterv);
1101 } catch (Exception e) {
1102 ci.println("Invalid format values: " + e.getMessage());
1106 if (pP <= 1 || fP <= 1 || dP <= 1 || tP <= 1) {
1107 ci.println("Invalid values. fP, pP, dP, tP have to be greater than 1.");
1111 statisticsTickNumber = fP;
1112 portTickNumber = pP;
1113 descriptionTickNumber = dP;
1114 tableTickNumber = tP;
1116 ci.println("New Values: fP=" + statisticsTickNumber + "s pP="
1117 + portTickNumber + "s dP=" + descriptionTickNumber + "s tP="
1118 + tableTickNumber + "s");
1122 * This method retrieves user configurations from config.ini and updates
1123 * statisticsTickNumber/portTickNumber/descriptionTickNumber accordingly.
1125 private void configStatsPollIntervals() {
1126 String fsStr = System.getProperty("of.flowStatsPollInterval");
1127 String psStr = System.getProperty("of.portStatsPollInterval");
1128 String dsStr = System.getProperty("of.descStatsPollInterval");
1129 String tsStr = System.getProperty("of.tableStatsPollInterval");
1130 Short fs, ps, ds, ts;
1132 if (fsStr != null) {
1134 fs = Short.parseShort(fsStr);
1136 statisticsTickNumber = fs;
1138 } catch (Exception e) {
1142 if (psStr != null) {
1144 ps = Short.parseShort(psStr);
1146 portTickNumber = ps;
1148 } catch (Exception e) {
1152 if (dsStr != null) {
1154 ds = Short.parseShort(dsStr);
1156 descriptionTickNumber = ds;
1158 } catch (Exception e) {
1162 if (tsStr != null) {
1164 ts = Short.parseShort(tsStr);
1166 tableTickNumber = ts;
1168 } catch (Exception e) {