-
/*
* Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
*
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import org.eclipse.osgi.framework.console.CommandInterpreter;
import org.eclipse.osgi.framework.console.CommandProvider;
import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
-import org.opendaylight.controller.protocol_plugin.openflow.IOFInventoryService;
+import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsListener;
import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6Match;
import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsReply;
import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsRequest;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
+import org.opendaylight.controller.sal.core.Node;
+import org.opendaylight.controller.sal.core.NodeConnector;
+import org.opendaylight.controller.sal.core.Property;
+import org.opendaylight.controller.sal.core.UpdateType;
+import org.opendaylight.controller.sal.utils.HexEncode;
import org.openflow.protocol.OFError;
import org.openflow.protocol.OFMatch;
import org.openflow.protocol.OFPort;
import org.openflow.protocol.OFStatisticsRequest;
import org.openflow.protocol.statistics.OFAggregateStatisticsRequest;
-import org.openflow.protocol.statistics.OFDescriptionStatistics;
import org.openflow.protocol.statistics.OFFlowStatisticsReply;
import org.openflow.protocol.statistics.OFFlowStatisticsRequest;
import org.openflow.protocol.statistics.OFPortStatisticsReply;
import org.openflow.protocol.statistics.OFQueueStatisticsRequest;
import org.openflow.protocol.statistics.OFStatistics;
import org.openflow.protocol.statistics.OFStatisticsType;
+import org.openflow.protocol.statistics.OFTableStatistics;
import org.openflow.protocol.statistics.OFVendorStatistics;
import org.openflow.util.HexString;
import org.osgi.framework.BundleContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.opendaylight.controller.sal.core.Name;
-import org.opendaylight.controller.sal.core.Node;
-import org.opendaylight.controller.sal.core.NodeConnector;
-import org.opendaylight.controller.sal.core.Property;
-import org.opendaylight.controller.sal.core.UpdateType;
-import org.opendaylight.controller.sal.utils.GlobalConstants;
-import org.opendaylight.controller.sal.utils.HexEncode;
-import org.opendaylight.controller.sal.utils.ServiceHelper;
-
/**
- * It periodically polls the different OF statistics from the OF switches
- * and caches them for quick retrieval for the above layers' modules
- * It also provides an API to directly query the switch about the statistics
- *
- *
- *
+ * Periodically polls the different OF statistics from the OF switches, caches
+ * them, and publishes results towards SAL. It also provides an API to directly
+ * query the switch for any specific statistics.
*/
-public class OFStatisticsManager implements IOFStatisticsManager,
- IInventoryShimExternalListener, CommandProvider {
- private static final Logger log = LoggerFactory
- .getLogger(OFStatisticsManager.class);
- private static final int initialSize = 64;
- private static final long flowStatsPeriod = 10000;
- private static final long descriptionStatsPeriod = 60000;
- private static final long portStatsPeriod = 5000;
- private long statisticsTimeout = 4000;
- private static final long tickPeriod = 1000;
- private static short statisticsTickNumber = (short) (flowStatsPeriod / tickPeriod);
- private static short descriptionTickNumber = (short) (descriptionStatsPeriod / tickPeriod);
- private static short portTickNumber = (short) (portStatsPeriod / tickPeriod);
+public class OFStatisticsManager implements IOFStatisticsManager, IInventoryShimExternalListener, CommandProvider {
+ private static final Logger log = LoggerFactory.getLogger(OFStatisticsManager.class);
+ private static final int INITIAL_SIZE = 64;
+ private static final long FLOW_STATS_PERIOD = 10000;
+ private static final long DESC_STATS_PERIOD = 60000;
+ private static final long PORT_STATS_PERIOD = 5000;
+ private static final long TABLE_STATS_PERIOD = 10000;
+ private static final long TICK = 1000;
+ private static short statisticsTickNumber = (short) (FLOW_STATS_PERIOD / TICK);
+ private static short descriptionTickNumber = (short) (DESC_STATS_PERIOD / TICK);
+ private static short portTickNumber = (short) (PORT_STATS_PERIOD / TICK);
+ private static short tableTickNumber = (short) (TABLE_STATS_PERIOD / TICK);
private static short factoredSamples = (short) 2;
private static short counter = 1;
private IController controller = null;
private ConcurrentMap<Long, List<OFStatistics>> flowStatistics;
private ConcurrentMap<Long, List<OFStatistics>> descStatistics;
private ConcurrentMap<Long, List<OFStatistics>> portStatistics;
- private List<OFStatistics> dummyList;
+ private ConcurrentMap<Long, List<OFStatistics>> tableStatistics;
private ConcurrentMap<Long, StatisticsTicks> statisticsTimerTicks;
protected BlockingQueue<StatsRequest> pendingStatsRequests;
protected BlockingQueue<Long> switchPortStatsUpdated;
private Timer statisticsTimer;
private TimerTask statisticsTimerTask;
private ConcurrentMap<Long, Boolean> switchSupportsVendorExtStats;
- private Map<Long, String> switchNamesDB;
- private Map<Long, Map<Short, TxRates>> txRates; // Per port sampled (every portStatsPeriod) transmit rate
+ // Per port sampled (every portStatsPeriod) transmit rate
+ private Map<Long, Map<Short, TxRates>> txRates;
+ private Set<IOFStatisticsListener> statisticsListeners = new CopyOnWriteArraySet<IOFStatisticsListener>();
/**
- * The object containing the latest factoredSamples tx rate samples
- * for a given switch port
+ * The object containing the latest factoredSamples tx rate samples for a
+ * given switch port
*/
protected class TxRates {
- Deque<Long> sampledTxBytes; // contains the latest factoredSamples sampled transmitted bytes
+ // contains the latest factoredSamples sampled transmitted bytes
+ Deque<Long> sampledTxBytes;
public TxRates() {
sampledTxBytes = new LinkedBlockingDeque<Long>();
public void update(Long txBytes) {
/*
- * Based on how many samples our average works on,
- * we might have to remove the oldest sample
+ * Based on how many samples our average works on, we might have to
+ * remove the oldest sample
*/
if (sampledTxBytes.size() == factoredSamples) {
sampledTxBytes.removeLast();
/**
* Returns the average transmit rate in bps
+ *
* @return the average transmit rate [bps]
*/
public long getAverageTxRate() {
if (sampledTxBytes.size() < factoredSamples) {
return average;
}
- long increment = (long) (sampledTxBytes.getFirst() - sampledTxBytes
- .getLast());
- long timePeriod = (long) (factoredSamples * portStatsPeriod)
- / (long) tickPeriod;
- average = (8 * increment) / timePeriod;
+ long increment = sampledTxBytes.getFirst() - sampledTxBytes
+ .getLast();
+ long timePeriod = factoredSamples * PORT_STATS_PERIOD / TICK;
+ average = (8L * increment) / timePeriod;
return average;
}
}
}
}
- /**
- * Function called by the dependency manager when all the required
- * dependencies are satisfied
- *
- */
- void init() {
+ private short getStatsQueueSize() {
+ String statsQueueSizeStr = System.getProperty("of.statsQueueSize");
+ short statsQueueSize = INITIAL_SIZE;
+ if (statsQueueSizeStr != null) {
+ try {
+ statsQueueSize = Short.parseShort(statsQueueSizeStr);
+ if (statsQueueSize <= 0) {
+ statsQueueSize = INITIAL_SIZE;
+ }
+ } catch (Exception e) {
+ }
+ }
+ return statsQueueSize;
}
- /**
- * Function called by the dependency manager when at least one
- * dependency become unsatisfied or when the component is shutting
- * down because for example bundle is being stopped.
- *
- */
- void destroy() {
+ IPluginOutConnectionService connectionPluginOutService;
+ void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+ connectionPluginOutService = s;
}
- /**
- * Function called by dependency manager after "init ()" is called
- * and after the services provided by the class are registered in
- * the service registry
- *
- */
- void start() {
- // Start managed timers
- statisticsTimer.scheduleAtFixedRate(statisticsTimerTask, 0, tickPeriod);
-
- // Start statistics collector thread
- statisticsCollector.start();
-
- // Start bandwidth utilization computer thread
- txRatesUpdater.start();
-
- // OSGI console
- registerWithOSGIConsole();
+ void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+ if (connectionPluginOutService == s) {
+ connectionPluginOutService = null;
+ }
}
/**
- * Function called by the dependency manager before the services
- * exported by the component are unregistered, this will be
- * followed by a "destroy ()" calls
+ * Function called by the dependency manager when all the required
+ * dependencies are satisfied
*
*/
- void stop() {
- // Stop managed timers
- statisticsTimer.cancel();
- }
-
- public OFStatisticsManager() {
+ void init() {
flowStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
descStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
portStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
- dummyList = new ArrayList<OFStatistics>(1);
- switchNamesDB = new HashMap<Long, String>();
- statisticsTimerTicks = new ConcurrentHashMap<Long, StatisticsTicks>(
- initialSize);
- pendingStatsRequests = new LinkedBlockingQueue<StatsRequest>(
- initialSize);
- switchPortStatsUpdated = new LinkedBlockingQueue<Long>(initialSize);
- switchSupportsVendorExtStats = new ConcurrentHashMap<Long, Boolean>(
- initialSize);
- txRates = new HashMap<Long, Map<Short, TxRates>>(initialSize);
+ tableStatistics = new ConcurrentHashMap<Long, List<OFStatistics>>();
+ pendingStatsRequests = new LinkedBlockingQueue<StatsRequest>(getStatsQueueSize());
+ statisticsTimerTicks = new ConcurrentHashMap<Long, StatisticsTicks>(INITIAL_SIZE);
+ switchPortStatsUpdated = new LinkedBlockingQueue<Long>(INITIAL_SIZE);
+ switchSupportsVendorExtStats = new ConcurrentHashMap<Long, Boolean>(INITIAL_SIZE);
+ txRates = new HashMap<Long, Map<Short, TxRates>>(INITIAL_SIZE);
+
+ configStatsPollIntervals();
// Initialize managed timers
- statisticsTimer = new Timer();
+ statisticsTimer = new Timer("Statistics Timer Ticks");
statisticsTimerTask = new TimerTask() {
@Override
public void run() {
while (true) {
try {
StatsRequest req = pendingStatsRequests.take();
- acquireStatistics(req.switchId, req.type,
- statisticsTimeout);
+ queryStatisticsInternal(req.switchId, req.type);
} catch (InterruptedException e) {
- log
- .warn("Flow Statistics Collector thread interrupted");
+ log.warn("Flow Statistics Collector thread "
+ + "interrupted", e);
+ return;
}
}
}
long switchId = switchPortStatsUpdated.take();
updatePortsTxRate(switchId);
} catch (InterruptedException e) {
- log.warn("TX Rate Updater thread interrupted");
+ log.warn("TX Rate Updater thread interrupted", e);
+ return;
}
}
}
}, "TX Rate Updater");
+ }
+ /**
+ * Function called by the dependency manager when at least one dependency
+ * become unsatisfied or when the component is shutting down because for
+ * example bundle is being stopped.
+ *
+ */
+ void destroy() {
+ statisticsListeners.clear();
+ }
+
+ /**
+ * Function called by dependency manager after "init ()" is called and after
+ * the services provided by the class are registered in the service registry
+ *
+ */
+ void start() {
+ // Start managed timers
+ statisticsTimer.scheduleAtFixedRate(statisticsTimerTask, 0, TICK);
+
+ // Start statistics collector thread
+ statisticsCollector.start();
+
+ // Start bandwidth utilization computer thread
+ txRatesUpdater.start();
+
+ // OSGI console
+ registerWithOSGIConsole();
+ }
+
+ /**
+ * Function called by the dependency manager before the services exported by
+ * the component are unregistered, this will be followed by a "destroy ()"
+ * calls
+ *
+ */
+ void stop() {
+ // Stop managed timers
+ statisticsTimer.cancel();
+ }
+
+ public void setStatisticsListener(IOFStatisticsListener s) {
+ this.statisticsListeners.add(s);
+ }
+
+ public void unsetStatisticsListener(IOFStatisticsListener s) {
+ if (s != null) {
+ this.statisticsListeners.remove(s);
+ }
}
private void registerWithOSGIConsole() {
- BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
- .getBundleContext();
- bundleContext.registerService(CommandProvider.class.getName(), this,
- null);
+ BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass()).getBundleContext();
+ bundleContext.registerService(CommandProvider.class.getName(), this, null);
}
private static class StatsRequest {
type = t;
}
+ @Override
public String toString() {
return "SReq = {switchId=" + switchId + ", type=" + type + "}";
}
}
private void addStatisticsTicks(Long switchId) {
- switchSupportsVendorExtStats.put(switchId, Boolean.TRUE); // Assume switch supports Vendor extension stats
+ switchSupportsVendorExtStats.put(switchId, Boolean.TRUE); // Assume
+ // switch
+ // supports
+ // Vendor
+ // extension
+ // stats
statisticsTimerTicks.put(switchId, new StatisticsTicks(true));
- log.info("Added Switch {} to target pool", HexString
- .toHexString(switchId.longValue()));
+ log.debug("Added Switch {} to target pool",
+ HexString.toHexString(switchId.longValue()));
}
protected static class StatisticsTicks {
private short flowStatisticsTicks;
private short descriptionTicks;
private short portStatisticsTicks;
+ private short tableStatisticsTicks;
public StatisticsTicks(boolean scattered) {
if (scattered) {
% statisticsTickNumber);
descriptionTicks = (short) (1 + counter % descriptionTickNumber);
portStatisticsTicks = (short) (1 + counter % portTickNumber);
+ tableStatisticsTicks = (short) (1 + counter % tableTickNumber);
} else {
flowStatisticsTicks = statisticsTickNumber;
descriptionTicks = descriptionTickNumber;
portStatisticsTicks = portTickNumber;
+ tableStatisticsTicks = tableTickNumber;
}
}
public boolean decrementFlowTicksIsZero() {
- // Please ensure no code is inserted between the if check and the flowStatisticsTicks reset
+ // Please ensure no code is inserted between the if check and the
+ // flowStatisticsTicks reset
if (--flowStatisticsTicks == 0) {
flowStatisticsTicks = statisticsTickNumber;
return true;
}
public boolean decrementDescTicksIsZero() {
- // Please ensure no code is inserted between the if check and the descriptionTicks reset
+ // Please ensure no code is inserted between the if check and the
+ // descriptionTicks reset
if (--descriptionTicks == 0) {
descriptionTicks = descriptionTickNumber;
return true;
}
public boolean decrementPortTicksIsZero() {
- // Please ensure no code is inserted between the if check and the descriptionTicks reset
+ // Please ensure no code is inserted between the if check and the
+ // descriptionTicks reset
if (--portStatisticsTicks == 0) {
portStatisticsTicks = portTickNumber;
return true;
return false;
}
+ public boolean decrementTableTicksIsZero() {
+ // Please ensure no code is inserted between the if check and the
+ // descriptionTicks reset
+ if(--tableStatisticsTicks == 0) {
+ tableStatisticsTicks = tableTickNumber;
+ return true;
+ }
+ return false;
+ }
+
+ @Override
public String toString() {
return "{fT=" + flowStatisticsTicks + ",dT=" + descriptionTicks
- + ",pT=" + portStatisticsTicks + "}";
+ + ",pT=" + portStatisticsTicks + ",tT=" + tableStatisticsTicks + "}";
}
}
private void printInfoMessage(String type, StatsRequest request) {
- log
- .info(
- type
- + " stats request not inserted for switch: {}. Queue size: {}. Collector state: {}.",
- new Object[] { HexString.toHexString(request.switchId),
- pendingStatsRequests.size(),
- statisticsCollector.getState().toString() });
+ log.trace("{} stats request not inserted for switch: {}. Queue size: {}. Collector state: {}.",
+ new Object[] {type, HexString.toHexString(request.switchId), pendingStatsRequests.size(),
+ statisticsCollector.getState().toString() });
}
protected void decrementTicks() {
.entrySet()) {
StatisticsTicks clock = entry.getValue();
Long switchId = entry.getKey();
- if (clock.decrementFlowTicksIsZero() == true) {
- request = (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) ? new StatsRequest(
- switchId, OFStatisticsType.VENDOR)
- : new StatsRequest(switchId, OFStatisticsType.FLOW);
- // If a request for this switch is already in the queue, skip to add this new request
+ if (clock.decrementFlowTicksIsZero()) {
+ request = (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) ?
+ new StatsRequest(switchId, OFStatisticsType.VENDOR) :
+ new StatsRequest(switchId, OFStatisticsType.FLOW);
+ // If a request for this switch is already in the queue, skip to
+ // add this new request
if (!pendingStatsRequests.contains(request)
&& false == pendingStatsRequests.offer(request)) {
printInfoMessage("Flow", request);
}
}
- if (clock.decrementDescTicksIsZero() == true) {
+ if (clock.decrementDescTicksIsZero()) {
request = new StatsRequest(switchId, OFStatisticsType.DESC);
- // If a request for this switch is already in the queue, skip to add this new request
+ // If a request for this switch is already in the queue, skip to
+ // add this new request
if (!pendingStatsRequests.contains(request)
&& false == pendingStatsRequests.offer(request)) {
printInfoMessage("Description", request);
}
}
- if (clock.decrementPortTicksIsZero() == true) {
+ if (clock.decrementPortTicksIsZero()) {
request = new StatsRequest(switchId, OFStatisticsType.PORT);
- // If a request for this switch is already in the queue, skip to add this new request
+ // If a request for this switch is already in the queue, skip to
+ // add this new request
if (!pendingStatsRequests.contains(request)
&& false == pendingStatsRequests.offer(request)) {
printInfoMessage("Port", request);
}
}
+
+ if(clock.decrementTableTicksIsZero()) {
+ request = new StatsRequest(switchId, OFStatisticsType.TABLE);
+ // If a request for this switch is already in the queue, skip to
+ // add this new request
+ if (!pendingStatsRequests.contains(request)
+ && false == pendingStatsRequests.offer(request)) {
+ printInfoMessage("Table", request);
+ }
+ }
}
}
private void removeStatsRequestTasks(Long switchId) {
- log.info("Cleaning Statistics database for switch "
- + HexEncode.longToHexString(switchId));
- // To be safe, let's attempt removal of both VENDOR and FLOW request. It does not hurt
+ log.debug("Cleaning Statistics database for switch {}",
+ HexEncode.longToHexString(switchId));
+ // To be safe, let's attempt removal of both VENDOR and FLOW request. It
+ // does not hurt
pendingStatsRequests.remove(new StatsRequest(switchId,
OFStatisticsType.VENDOR));
pendingStatsRequests.remove(new StatsRequest(switchId,
OFStatisticsType.DESC));
pendingStatsRequests.remove(new StatsRequest(switchId,
OFStatisticsType.PORT));
+ pendingStatsRequests.remove(new StatsRequest(switchId,
+ OFStatisticsType.TABLE));
// Take care of the TX rate databases
switchPortStatsUpdated.remove(switchId);
txRates.remove(switchId);
statisticsTimerTicks.remove(switchId);
removeStatsRequestTasks(switchId);
flowStatistics.remove(switchId);
- log.info("Statistics removed for switch "
- + HexString.toHexString(switchId));
+ log.debug("Statistics removed for switch {}",
+ HexString.toHexString(switchId));
}
- private void acquireStatistics(Long switchId, OFStatisticsType statType,
- long timeout) {
+ private void queryStatisticsInternal(Long switchId, OFStatisticsType statType) {
// Query the switch on all matches
- List<OFStatistics> values = this.acquireStatistics(switchId, statType,
- null, timeout);
-
- // Update local caching database if got a valid response
- if (values != null && !values.isEmpty()) {
- if ((statType == OFStatisticsType.FLOW)
- || (statType == OFStatisticsType.VENDOR)) {
- flowStatistics.put(switchId, values);
- } else if (statType == OFStatisticsType.DESC) {
- if ((switchNamesDB != null)
- && switchNamesDB.containsKey(switchId)) {
- // Check if user manually configured a name for the switch
- for (OFStatistics entry : values) {
- OFDescriptionStatistics reply = (OFDescriptionStatistics) entry;
- reply.setSerialNumber(switchNamesDB.get(switchId));
- }
- }
- // check if notification is needed
- if (descStatistics.get(switchId) == null
- || !(descStatistics.get(switchId).get(0).equals(values
- .get(0)))) {
- IOFInventoryService inventory = (IOFInventoryService) ServiceHelper
- .getInstance(IOFInventoryService.class,
- GlobalConstants.DEFAULT.toString(), this);
- if (inventory != null) {
- // Notify Inventory Service about the name update
- Set<Property> propSet = new HashSet<Property>(1);
- Name name = new Name(((OFDescriptionStatistics) values
- .get(0)).getSerialNumber());
- propSet.add(name);
- inventory.updateSwitchProperty(switchId, propSet);
- }
- }
- // overwrite cache
- descStatistics.put(switchId, values);
- } else if (statType == OFStatisticsType.PORT) {
- // Overwrite cache with new port statistics for this switch
- portStatistics.put(switchId, values);
-
- // Wake up the thread which maintains the TX byte counters for each port
- switchPortStatsUpdated.offer(switchId);
+ List<OFStatistics> values = this.fetchStatisticsFromSwitch(switchId, statType, null);
+
+ // If got a valid response update local cache and notify listeners
+ if (!values.isEmpty()) {
+ switch (statType) {
+ case FLOW:
+ case VENDOR:
+ flowStatistics.put(switchId, values);
+ notifyFlowUpdate(switchId, values);
+ break;
+ case DESC:
+ // Overwrite cache
+ descStatistics.put(switchId, values);
+ // Notify who may be interested in a description change
+ notifyDescriptionUpdate(switchId, values);
+ break;
+ case PORT:
+ // Overwrite cache with new port statistics for this switch
+ portStatistics.put(switchId, values);
+
+ // Wake up the thread which maintains the TX byte counters for
+ // each port
+ switchPortStatsUpdated.offer(switchId);
+ notifyPortUpdate(switchId, values);
+ break;
+ case TABLE:
+ // Overwrite cache
+ tableStatistics.put(switchId, values);
+ notifyTableUpdate(switchId, values);
+ break;
+ default:
}
}
}
+ private void notifyDescriptionUpdate(Long switchId, List<OFStatistics> values) {
+ for (IOFStatisticsListener l : this.statisticsListeners) {
+ l.descriptionStatisticsRefreshed(switchId, values);
+ }
+ }
+
+ private void notifyFlowUpdate(Long switchId, List<OFStatistics> values) {
+ if (values.get(0) instanceof OFVendorStatistics) {
+ values = this.v6StatsListToOFStatsList(values);
+ }
+
+ for (IOFStatisticsListener l : this.statisticsListeners) {
+ l.flowStatisticsRefreshed(switchId, values);
+ }
+
+ }
+
+ private void notifyPortUpdate(Long switchId, List<OFStatistics> values) {
+ for (IOFStatisticsListener l : this.statisticsListeners) {
+ l.portStatisticsRefreshed(switchId, values);
+ }
+ }
+
+ private void notifyTableUpdate(Long switchId, List<OFStatistics> values) {
+ for (IOFStatisticsListener l : this.statisticsListeners) {
+ l.tableStatisticsRefreshed(switchId, values);
+ }
+ }
+
/*
- * Generic function to get the statistics form a OF switch
+ * Generic function to get the statistics form an OF switch
*/
@SuppressWarnings("unchecked")
- private List<OFStatistics> acquireStatistics(Long switchId,
- OFStatisticsType statsType, Object target, long timeout) {
- List<OFStatistics> values = null;
+ private List<OFStatistics> fetchStatisticsFromSwitch(Long switchId,
+ OFStatisticsType statsType, Object target) {
+ List<OFStatistics> values = Collections.emptyList();
String type = null;
ISwitch sw = controller.getSwitch(switchId);
match.setWildcards(0xffffffff);
} else if (!(target instanceof OFMatch)) {
// Malformed request
- log.warn("Invalid target type for Flow stats request: "
- + target.getClass());
- return null;
+ log.warn("Invalid target type for Flow stats request: {}",
+ target.getClass());
+ return Collections.emptyList();
} else {
// Specific flow request
match = (OFMatch) target;
short targetPort;
if (target == null) {
// All ports request
- targetPort = (short) OFPort.OFPP_NONE.getValue();
+ targetPort = OFPort.OFPP_NONE.getValue();
} else if (!(target instanceof Short)) {
// Malformed request
- log.warn("Invalid target type for Port stats request: "
- + target.getClass());
- return null;
+ log.warn("Invalid target type for Port stats request: {}",
+ target.getClass());
+ return Collections.emptyList();
} else {
// Specific port request
targetPort = (Short) target;
type = "PORT";
} else if (statsType == OFStatisticsType.QUEUE) {
OFQueueStatisticsRequest specificReq = new OFQueueStatisticsRequest();
- specificReq.setPortNumber((short) OFPort.OFPP_ALL.getValue());
+ specificReq.setPortNumber(OFPort.OFPP_ALL.getValue());
specificReq.setQueueId(0xffffffff);
req.setStatistics(Collections
.singletonList((OFStatistics) specificReq));
} else if (statsType == OFStatisticsType.DESC) {
type = "DESC";
} else if (statsType == OFStatisticsType.TABLE) {
+ if(target != null){
+ if (!(target instanceof Byte)) {
+ // Malformed request
+ log.warn("Invalid table id for table stats request: {}",
+ target.getClass());
+ return Collections.emptyList();
+ }
+ byte targetTable = (Byte) target;
+ OFTableStatistics specificReq = new OFTableStatistics();
+ specificReq.setTableId(targetTable);
+ req.setStatistics(Collections
+ .singletonList((OFStatistics) specificReq));
+ requestLength += specificReq.getLength();
+ }
type = "TABLE";
}
req.setLengthU(requestLength);
log.warn("Request Timed Out for ({}) from switch {}", type,
HexString.toHexString(switchId));
} else if (result instanceof OFError) {
- log.warn("Switch {} failed to handle ({}) stats request: "
- + Utils.getOFErrorString((OFError) result), HexString
- .toHexString(switchId), type);
+ log.warn("Switch {} failed to handle ({}) stats request: {}",
+ new Object[] { HexString.toHexString(switchId), type,
+ Utils.getOFErrorString((OFError) result) });
if (this.switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) {
- log
- .warn(
- "Switching back to regular Flow stats requests for switch {}",
- HexString.toHexString(switchId));
+ log.warn(
+ "Switching back to regular Flow stats requests for switch {}",
+ HexString.toHexString(switchId));
this.switchSupportsVendorExtStats.put(switchId,
Boolean.FALSE);
}
List<OFStatistics> list = flowStatistics.get(switchId);
/*
- * Check on emptiness as interference between add and get is still
- * possible on the inner list (the concurrentMap entry's value)
+ * Check on emptiness as interference between add and get is still
+ * possible on the inner list (the concurrentMap entry's value)
*/
- return (list == null || list.isEmpty()) ? this.dummyList
+ return (list == null || list.isEmpty()) ? Collections.<OFStatistics>emptyList()
: (list.get(0) instanceof OFVendorStatistics) ? this
.v6StatsListToOFStatsList(list) : list;
}
@Override
- public List<OFStatistics> getOFFlowStatistics(Long switchId, OFMatch ofMatch) {
+ public List<OFStatistics> getOFFlowStatistics(Long switchId, OFMatch ofMatch, short priority) {
List<OFStatistics> statsList = flowStatistics.get(switchId);
/*
- * Check on emptiness as interference between add and get is still
- * possible on the inner list (the concurrentMap entry's value)
+ * Check on emptiness as interference between add and get is still
+ * possible on the inner list (the concurrentMap entry's value)
*/
if (statsList == null || statsList.isEmpty()) {
- return this.dummyList;
+ return Collections.emptyList();
}
if (statsList.get(0) instanceof OFVendorStatistics) {
/*
- * Caller could provide regular OF match when we
- * instead pull the vendor statistics from this node
- * Caller is not supposed to know whether this switch supports
- * vendor extensions statistics requests
+ * Caller could provide regular OF match when we instead pull the
+ * vendor statistics from this node Caller is not supposed to know
+ * whether this switch supports vendor extensions statistics
+ * requests
*/
V6Match targetMatch = (ofMatch instanceof V6Match) ? (V6Match) ofMatch
: new V6Match(ofMatch);
for (OFStatistics stats : targetList) {
V6StatsReply v6Stats = (V6StatsReply) stats;
V6Match v6Match = v6Stats.getMatch();
- if (v6Match.equals(targetMatch)) {
+ if (v6Stats.getPriority() == priority && targetMatch.equals(v6Match)) {
List<OFStatistics> list = new ArrayList<OFStatistics>();
list.add(stats);
return list;
} else {
for (OFStatistics stats : statsList) {
OFFlowStatisticsReply flowStats = (OFFlowStatisticsReply) stats;
- if (flowStats.getMatch().equals(ofMatch)) {
+ if (flowStats.getPriority() == priority && ofMatch.equals(flowStats.getMatch())) {
List<OFStatistics> list = new ArrayList<OFStatistics>();
list.add(stats);
return list;
}
}
}
- return this.dummyList;
+ return Collections.emptyList();
}
/*
* Converts the v6 vendor statistics to the OFStatistics
*/
- private List<OFStatistics> v6StatsListToOFStatsList(
- List<OFStatistics> statistics) {
+ private List<OFStatistics> v6StatsListToOFStatsList(List<OFStatistics> statistics) {
+ if (statistics == null || statistics.isEmpty()) {
+ return Collections.emptyList();
+ }
List<OFStatistics> v6statistics = new ArrayList<OFStatistics>();
- if (statistics != null && !statistics.isEmpty()) {
- for (OFStatistics stats : statistics) {
- if (stats instanceof OFVendorStatistics) {
- List<OFStatistics> r = getV6ReplyStatistics((OFVendorStatistics) stats);
- if (r != null) {
- v6statistics.addAll(r);
- }
+ for (OFStatistics stats : statistics) {
+ if (stats instanceof OFVendorStatistics) {
+ List<OFStatistics> r = getV6ReplyStatistics((OFVendorStatistics) stats);
+ if (r != null) {
+ v6statistics.addAll(r);
}
}
}
OFVendorStatistics stat) {
int length = stat.getLength();
List<OFStatistics> results = new ArrayList<OFStatistics>();
- if (length < 12)
- return null; // Nicira Hdr is 12 bytes. We need atleast that much
+ if (length < 12) {
+ // Nicira Hdr is 12 bytes. We need at least that much
+ return Collections.emptyList();
+ }
ByteBuffer data = ByteBuffer.allocate(length);
stat.writeTo(data);
data.rewind();
- log.trace("getV6ReplyStatistics: Buffer BYTES ARE {}", HexString
- .toHexString(data.array()));
+ if (log.isTraceEnabled()) {
+ log.trace("getV6ReplyStatistics: Buffer BYTES ARE {}",
+ HexString.toHexString(data.array()));
+ }
- int vendor = data.getInt(); //first 4 bytes is vendor id.
+ int vendor = data.getInt(); // first 4 bytes is vendor id.
if (vendor != V6StatsRequest.NICIRA_VENDOR_ID) {
- log
- .debug("Unexpected vendor id: 0x{}", Integer
- .toHexString(vendor));
- return null;
+ log.warn("Unexpected vendor id: 0x{}", Integer.toHexString(vendor));
+ return Collections.emptyList();
} else {
- //go ahead by 8 bytes which is 8 bytes of 0
- data.getLong(); //should be all 0's
- length -= 12; // 4 bytes Nicira Hdr + 8 bytes from above line have been consumed
+ // go ahead by 8 bytes which is 8 bytes of 0
+ data.getLong(); // should be all 0's
+ length -= 12; // 4 bytes Nicira Hdr + 8 bytes from above line have
+ // been consumed
}
V6StatsReply v6statsreply;
while (length > 0) {
v6statsreply = new V6StatsReply();
min_len = v6statsreply.getLength();
- if (length < v6statsreply.getLength())
+ if (length < v6statsreply.getLength()) {
break;
+ }
v6statsreply.setActionFactory(stat.getActionFactory());
v6statsreply.readFrom(data);
- if (v6statsreply.getLength() < min_len)
+ if (v6statsreply.getLength() < min_len) {
break;
+ }
v6statsreply.setVendorId(vendor);
log.trace("V6StatsReply: {}", v6statsreply);
length -= v6statsreply.getLength();
@Override
public List<OFStatistics> queryStatistics(Long switchId,
- OFStatisticsType statType, Object target, long timeout) {
+ OFStatisticsType statType, Object target) {
/*
- * Caller does not know and it is not supposed to know whether
- * this switch supports vendor extension. We adjust the target for him
+ * Caller does not know and it is not supposed to know whether this
+ * switch supports vendor extension. We adjust the target for him
*/
if (statType == OFStatisticsType.FLOW) {
if (switchSupportsVendorExtStats.get(switchId) == Boolean.TRUE) {
}
}
- List<OFStatistics> list = this.acquireStatistics(switchId, statType,
- target, timeout);
+ List<OFStatistics> list = this.fetchStatisticsFromSwitch(switchId, statType, target);
- return (list == null) ? null
- : (statType == OFStatisticsType.VENDOR) ? v6StatsListToOFStatsList(list)
- : list;
+ return (statType == OFStatisticsType.VENDOR) ? v6StatsListToOFStatsList(list) : list;
}
@Override
public List<OFStatistics> getOFDescStatistics(Long switchId) {
- if (!descStatistics.containsKey(switchId))
- return this.dummyList;
+ if (!descStatistics.containsKey(switchId)) {
+ return Collections.emptyList();
+ }
return descStatistics.get(switchId);
}
@Override
public List<OFStatistics> getOFPortStatistics(Long switchId) {
if (!portStatistics.containsKey(switchId)) {
- return this.dummyList;
+ return Collections.emptyList();
}
return portStatistics.get(switchId);
@Override
public List<OFStatistics> getOFPortStatistics(Long switchId, short portId) {
if (!portStatistics.containsKey(switchId)) {
- return this.dummyList;
+ return Collections.emptyList();
}
List<OFStatistics> list = new ArrayList<OFStatistics>(1);
for (OFStatistics stats : portStatistics.get(switchId)) {
return list;
}
+ @Override
+ public List<OFStatistics> getOFTableStatistics(Long switchId) {
+ if (!tableStatistics.containsKey(switchId)) {
+ return Collections.emptyList();
+ }
+
+ return tableStatistics.get(switchId);
+ }
+
+ @Override
+ public List<OFStatistics> getOFTableStatistics(Long switchId, Byte tableId) {
+ if (!tableStatistics.containsKey(switchId)) {
+ return Collections.emptyList();
+ }
+
+ List<OFStatistics> list = new ArrayList<OFStatistics>(1);
+ for (OFStatistics stats : tableStatistics.get(switchId)) {
+ if (((OFTableStatistics) stats).getTableId() == tableId) {
+ list.add(stats);
+ break;
+ }
+ }
+ return list;
+ }
+
@Override
public int getFlowsNumber(long switchId) {
return this.flowStatistics.get(switchId).size();
}
/*
- * InventoryShim replay for us all the switch addition which happened before we were brought up
+ * InventoryShim replay for us all the switch addition which happened before
+ * we were brought up
*/
@Override
public void updateNode(Node node, UpdateType type, Set<Property> props) {
}
/**
- * Update the cached port rates for this switch with the latest
- * retrieved port transmit byte count
+ * Update the cached port rates for this switch with the latest retrieved
+ * port transmit byte count
+ *
* @param switchId
*/
private synchronized void updatePortsTxRate(long switchId) {
- List<OFStatistics> newPortStatistics = this.portStatistics
- .get(switchId);
+ List<OFStatistics> newPortStatistics = this.portStatistics.get(switchId);
if (newPortStatistics == null) {
return;
}
@Override
public String getHelp() {
StringBuffer help = new StringBuffer();
- help.append("---OF Switch ID to Name Mapping---\n");
- help
- .append("\t ofaddname - Add a switchId to name mapping entry\n");
- help
- .append("\t ofdeletename - Delete a switchId from the mapping db\n");
- help.append("\t ofprintnames - Print the mapping db\n");
help.append("---OF Statistics Manager utilities---\n");
- help
- .append("\t ofdumpstatsmgr - Print Internal Stats Mgr db\n");
- help
- .append("\t ofstatstimeout - Change Statistics request's timeout value\n");
+ help.append("\t ofdumpstatsmgr - "
+ + "Print Internal Stats Mgr db\n");
+ help.append("\t ofstatsmgrintervals <fP> <pP> <dP> <tP> (all in seconds) - "
+ + "Set/Show flow/port/dedscription stats poll intervals\n");
return help.toString();
}
return Long.parseLong(switchString, radix);
}
- public void _ofaddname(CommandInterpreter ci) {
- if (switchNamesDB == null)
- switchNamesDB = new HashMap<Long, String>();
- String switchId = ci.nextArgument();
- if (!isValidSwitchId(switchId)) {
- ci.println("Please provide a valid SwithcId");
- return;
- }
- Long sid = getSwitchIDLong(switchId);
- String switchName = ci.nextArgument();
- if (switchName == null) {
- ci.println("Please provide a valid Switch name");
- return;
- }
- switchNamesDB.put(sid, switchName);
- }
-
- public void _ofdeletename(CommandInterpreter ci) {
- if (switchNamesDB == null)
- return;
- String switchId = ci.nextArgument();
- if (!isValidSwitchId(switchId)) {
- ci.println("Please provide a valid SwitchId");
- return;
- }
- Long sid = getSwitchIDLong(switchId);
- switchNamesDB.remove(sid);
- }
-
- public void _ofprintnames(CommandInterpreter ci) {
- if (switchNamesDB == null)
- return;
- for (Long key : switchNamesDB.keySet()) {
- ci.println(key + " -> " + switchNamesDB.get(key) + "\n");
- }
- }
-
/*
* Internal information dump code
*/
public void _ofdumpstatsmgr(CommandInterpreter ci) {
ci.println("Global Counter: " + counter);
- ci
- .println("Timer Ticks: "
- + prettyPrintSwitchMap(statisticsTimerTicks));
+ ci.println("Timer Ticks: " + prettyPrintSwitchMap(statisticsTimerTicks));
ci.println("PendingStatsQueue: " + pendingStatsRequests);
ci.println("PendingStatsQueue size: " + pendingStatsRequests.size());
ci.println("Stats Collector alive: " + statisticsCollector.isAlive());
ci.println("Flow Stats Period: " + statisticsTickNumber + " s");
ci.println("Desc Stats Period: " + descriptionTickNumber + " s");
ci.println("Port Stats Period: " + portTickNumber + " s");
- ci.println("Stats request timeout: " + Float.valueOf(statisticsTimeout)
- / 1000.0 + " s");
- }
-
- public void _ofstatstimeout(CommandInterpreter ci) {
- String timeoutString = ci.nextArgument();
- if (timeoutString == null || !timeoutString.matches("^[0-9]+$")) {
- ci.println("Invalid value. Has to be numeric.");
- return;
- }
-
- long newTimeout = Long.valueOf(timeoutString);
- if (newTimeout < 50 || newTimeout > 60000) {
- ci.println("Invalid value. Valid range is [50-60000]ms");
- return;
- }
- this.statisticsTimeout = newTimeout;
- ci.println("New value: " + statisticsTimeout + " ms");
+ ci.println("Table Stats Period: " + tableTickNumber + " s");
}
public void _resetSwitchCapability(CommandInterpreter ci) {
String averageWindow = ci.nextArgument();
short seconds = 0;
if (averageWindow == null) {
- ci
- .println("Insert the length in seconds of the average window for tx rate");
- ci
- .println("Current: " + factoredSamples * portTickNumber
- + " secs");
+ ci.println("Insert the length in seconds of the median "
+ + "window for tx rate");
+ ci.println("Current: " + factoredSamples * portTickNumber + " secs");
return;
}
try {
ci.println("New: " + factoredSamples * portTickNumber + " secs");
}
+ public void _ofstatsmgrintervals(CommandInterpreter ci) {
+ String flowStatsInterv = ci.nextArgument();
+ String portStatsInterv = ci.nextArgument();
+ String descStatsInterv = ci.nextArgument();
+ String tableStatsInterv = ci.nextArgument();
+
+ if (flowStatsInterv == null || portStatsInterv == null
+ || descStatsInterv == null) {
+ ci.println("Usage: ofstatsmgrintervals <fP> <pP> <dP> <tP> (all in seconds)");
+ ci.println("Current Values: fP=" + statisticsTickNumber + "sec pP="
+ + portTickNumber + "sec dP=" + descriptionTickNumber + "sec tP=" + tableTickNumber + " sec");
+ return;
+ }
+ Short fP, pP, dP, tP;
+ try {
+ fP = Short.parseShort(flowStatsInterv);
+ pP = Short.parseShort(portStatsInterv);
+ dP = Short.parseShort(descStatsInterv);
+ tP = Short.parseShort(tableStatsInterv);
+ } catch (Exception e) {
+ ci.println("Invalid format values: " + e.getMessage());
+ return;
+ }
+
+ if (pP <= 1 || fP <= 1 || dP <= 1 || tP <= 1) {
+ ci.println("Invalid values. fP, pP, dP, tP have to be greater than 1.");
+ return;
+ }
+
+ statisticsTickNumber = fP;
+ portTickNumber = pP;
+ descriptionTickNumber = dP;
+ tableTickNumber = tP;
+
+ ci.println("New Values: fP=" + statisticsTickNumber + "s pP="
+ + portTickNumber + "s dP=" + descriptionTickNumber + "s tP="
+ + tableTickNumber + "s");
+ }
+
+ /**
+ * This method retrieves user configurations from config.ini and updates
+ * statisticsTickNumber/portTickNumber/descriptionTickNumber accordingly.
+ */
+ private void configStatsPollIntervals() {
+ String fsStr = System.getProperty("of.flowStatsPollInterval");
+ String psStr = System.getProperty("of.portStatsPollInterval");
+ String dsStr = System.getProperty("of.descStatsPollInterval");
+ String tsStr = System.getProperty("of.tableStatsPollInterval");
+ Short fs, ps, ds, ts;
+
+ if (fsStr != null) {
+ try {
+ fs = Short.parseShort(fsStr);
+ if (fs > 0) {
+ statisticsTickNumber = fs;
+ }
+ } catch (Exception e) {
+ }
+ }
+
+ if (psStr != null) {
+ try {
+ ps = Short.parseShort(psStr);
+ if (ps > 0) {
+ portTickNumber = ps;
+ }
+ } catch (Exception e) {
+ }
+ }
+
+ if (dsStr != null) {
+ try {
+ ds = Short.parseShort(dsStr);
+ if (ds > 0) {
+ descriptionTickNumber = ds;
+ }
+ } catch (Exception e) {
+ }
+ }
+
+ if (tsStr != null) {
+ try{
+ ts = Short.parseShort(tsStr);
+ if (ts > 0) {
+ tableTickNumber = ts;
+ }
+ } catch (Exception e) {
+ }
+ }
+ }
}