1 package org.opendaylight.controller.md.statistics.manager.impl;
3 import java.util.Collections;
4 import java.util.HashMap;
7 import java.util.Map.Entry;
8 import java.util.concurrent.ExecutorService;
9 import java.util.concurrent.Executors;
10 import java.util.concurrent.ThreadFactory;
12 import org.opendaylight.controller.md.statistics.manager.StatPermCollector;
13 import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
14 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
15 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
16 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId;
17 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
18 import org.slf4j.Logger;
19 import org.slf4j.LoggerFactory;
21 import com.google.common.base.Preconditions;
22 import com.google.common.util.concurrent.ThreadFactoryBuilder;
26 * org.opendaylight.controller.md.statistics.manager.impl
28 * StatPermCollectorImpl
29 * Thread base statistic collector. Class holds internal map for all registered
30 * (means connected) nodes with List of Switch capabilities;
31 * Statistics collecting process get cross whole Network Device by device
32 * and statistic by statistic (follow Switch capabilities to prevent unnecessary
33 * ask) Next statistic start collecting by notification or by timeout.
35 * @author @author avishnoi@in.ibm.com <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
38 public class StatPermCollectorImpl implements StatPermCollector {
40 private final static Logger LOG = LoggerFactory.getLogger(StatPermCollectorImpl.class);
42 private final static long STAT_COLLECT_TIME_OUT = 30000L;
44 private final ExecutorService statNetCollectorServ;
45 private final StatisticsManager manager;
47 private final int maxNodeForCollector;
48 private final long minReqNetInterval;
49 private final String name;
51 private final Object statCollectorLock = new Object();
52 private final Object statNodeHolderLock = new Object();
54 private Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNodeHolder =
55 Collections.<InstanceIdentifier<Node>, StatNodeInfoHolder> emptyMap();
57 private volatile boolean wakeMe = false;
58 private volatile boolean finishing = false;
60 public StatPermCollectorImpl(final StatisticsManager manager, final long minReqNetInterv, final int nr,
61 final int maxNodeForCollectors) {
62 this.manager = Preconditions.checkNotNull(manager, "StatisticsManager can not be null!");
63 name = "odl-stat-collector-" + nr;
64 minReqNetInterval = minReqNetInterv;
65 final ThreadFactory threadFact = new ThreadFactoryBuilder()
66 .setNameFormat(name + "-thread-%d").build();
67 statNetCollectorServ = Executors.newSingleThreadExecutor(threadFact);
68 maxNodeForCollector = maxNodeForCollectors;
69 LOG.trace("StatCollector {} start successfull!", name);
73 * finish collecting statistics
77 statNodeHolder = Collections.<InstanceIdentifier<Node>, StatNodeInfoHolder> emptyMap();
79 collectNextStatistics();
80 statNetCollectorServ.shutdown();
84 public boolean isProvidedFlowNodeActive(
85 final InstanceIdentifier<Node> flowNode) {
86 return statNodeHolder.containsKey(flowNode);
90 public boolean connectedNodeRegistration(final InstanceIdentifier<Node> ident,
91 final List<StatCapabTypes> statTypes, final Short nrOfSwitchTables) {
92 if (ident.isWildcarded()) {
93 LOG.warn("FlowCapableNode IstanceIdentifier {} registration can not be wildcarded!", ident);
95 if ( ! statNodeHolder.containsKey(ident)) {
96 synchronized (statNodeHolderLock) {
97 final boolean startStatCollecting = statNodeHolder.size() == 0;
98 if ( ! statNodeHolder.containsKey(ident)) {
99 if (statNodeHolder.size() >= maxNodeForCollector) {
102 final Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNode =
103 new HashMap<>(statNodeHolder);
104 final NodeRef nodeRef = new NodeRef(ident);
105 final StatNodeInfoHolder nodeInfoHolder = new StatNodeInfoHolder(nodeRef,
106 statTypes, nrOfSwitchTables);
107 statNode.put(ident, nodeInfoHolder);
108 statNodeHolder = Collections.unmodifiableMap(statNode);
110 if (startStatCollecting) {
112 statNetCollectorServ.execute(this);
121 public boolean disconnectedNodeUnregistration(final InstanceIdentifier<Node> ident) {
122 if (ident.isWildcarded()) {
123 LOG.warn("FlowCapableNode IstanceIdentifier {} unregistration can not be wildcarded!", ident);
125 if (statNodeHolder.containsKey(ident)) {
126 synchronized (statNodeHolderLock) {
127 if (statNodeHolder.containsKey(ident)) {
128 final Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNode =
129 new HashMap<>(statNodeHolder);
130 statNode.remove(ident);
131 statNodeHolder = Collections.unmodifiableMap(statNode);
133 if (statNodeHolder.isEmpty()) {
135 collectNextStatistics();
136 statNetCollectorServ.shutdown();
146 public void collectNextStatistics() {
148 synchronized (statCollectorLock) {
150 LOG.trace("STAT-COLLECTOR is notified to conntinue");
151 statCollectorLock.notify();
162 catch (final InterruptedException e1) {
165 LOG.debug("StatCollector {} Start collecting!", name);
166 /* Neverending cyle - wait for finishing */
167 while ( ! finishing) {
168 boolean collecting = false;
169 final long startTime = System.currentTimeMillis();
171 if ( ! statNodeHolder.isEmpty()) {
173 collectStatCrossNetwork();
178 final long statFinalTime = System.currentTimeMillis() - startTime;
179 LOG.debug("STAT-MANAGER {}: last all NET statistics collection cost {} ms", name, statFinalTime);
180 if (statFinalTime < minReqNetInterval) {
181 LOG.trace("statCollector is about to make a collecting sleep");
182 synchronized (statCollectorLock) {
185 final long waitTime = minReqNetInterval - statFinalTime;
186 statCollectorLock.wait(waitTime);
187 LOG.trace("STAT-MANAGER : statCollector {} is waking up from a collecting sleep for {} ms", name, waitTime);
188 } catch (final InterruptedException e) {
189 LOG.warn("statCollector has been interrupted during collecting sleep", e);
199 private void waitingForNotification() {
200 synchronized (statCollectorLock) {
203 statCollectorLock.wait(STAT_COLLECT_TIME_OUT);
204 LOG.trace("statCollector is waking up from a wait stat Response sleep");
205 } catch (final InterruptedException e) {
206 LOG.warn("statCollector has been interrupted waiting stat Response sleep", e);
214 private void collectStatCrossNetwork() {
215 for (final Entry<InstanceIdentifier<Node>, StatNodeInfoHolder> nodeEntity : statNodeHolder.entrySet()) {
216 final List<StatCapabTypes> listNeededStat = nodeEntity.getValue().getStatMarkers();
217 final NodeRef actualNodeRef = nodeEntity.getValue().getNodeRef();
218 final Short maxTables = nodeEntity.getValue().getMaxTables();
219 for (final StatCapabTypes statMarker : listNeededStat) {
220 if ( ! isProvidedFlowNodeActive(nodeEntity.getKey())) {
223 switch (statMarker) {
225 LOG.trace("STAT-MANAGER-collecting PORT-STATS for NodeRef {}", actualNodeRef);
226 manager.getRpcMsgManager().getAllPortsStat(actualNodeRef);
227 waitingForNotification();
230 LOG.trace("STAT-MANAGER-collecting QUEUE-STATS for NodeRef {}", actualNodeRef);
231 manager.getRpcMsgManager().getAllQueueStat(actualNodeRef);
232 waitingForNotification();
235 LOG.trace("STAT-MANAGER-collecting TABLE-STATS for NodeRef {}", actualNodeRef);
236 manager.getRpcMsgManager().getAllTablesStat(actualNodeRef);
237 waitingForNotification();
240 LOG.trace("STAT-MANAGER-collecting GROUP-STATS for NodeRef {}", actualNodeRef);
241 manager.getRpcMsgManager().getGroupFeaturesStat(actualNodeRef);
242 waitingForNotification();
243 manager.getRpcMsgManager().getAllGroupsConfStats(actualNodeRef);
244 waitingForNotification();
245 manager.getRpcMsgManager().getAllGroupsStat(actualNodeRef);
246 waitingForNotification();
249 LOG.trace("STAT-MANAGER-collecting METER-STATS for NodeRef {}", actualNodeRef);
250 manager.getRpcMsgManager().getMeterFeaturesStat(actualNodeRef);
251 waitingForNotification();
252 manager.getRpcMsgManager().getAllMeterConfigStat(actualNodeRef);
253 waitingForNotification();
254 manager.getRpcMsgManager().getAllMetersStat(actualNodeRef);
255 waitingForNotification();
258 LOG.trace("STAT-MANAGER-collecting FLOW-STATS-ALL_FLOWS for NodeRef {}", actualNodeRef);
259 manager.getRpcMsgManager().getAllFlowsStat(actualNodeRef);
260 waitingForNotification();
261 LOG.trace("STAT-MANAGER-collecting FLOW-AGGREGATE-STATS for NodeRef {}", actualNodeRef);
262 for (short i = 0; i < maxTables; i++) {
263 final TableId tableId = new TableId(i);
264 manager.getRpcMsgManager().getAggregateFlowStat(actualNodeRef, tableId);
268 /* Exception for programmers in implementation cycle */
269 throw new IllegalStateException("Not implemented ASK for " + statMarker);
275 private class StatNodeInfoHolder {
276 private final NodeRef nodeRef;
277 private final List<StatCapabTypes> statMarkers;
278 private final Short maxTables;
280 public StatNodeInfoHolder(final NodeRef nodeRef,
281 final List<StatCapabTypes> statMarkers, final Short maxTables) {
282 this.nodeRef = nodeRef;
283 this.maxTables = maxTables;
284 this.statMarkers = statMarkers;
287 public final NodeRef getNodeRef() {
291 public final List<StatCapabTypes> getStatMarkers() {
295 public final Short getMaxTables() {
301 public boolean hasActiveNodes() {
302 return ( ! statNodeHolder.isEmpty());