Fix bug 2413 NPE for group and meters
[controller.git] / opendaylight / md-sal / statistics-manager / src / main / java / org / opendaylight / controller / md / statistics / manager / impl / StatPermCollectorImpl.java
1 package org.opendaylight.controller.md.statistics.manager.impl;
2
3 import java.util.ArrayList;
4 import java.util.Collections;
5 import java.util.HashMap;
6 import java.util.List;
7 import java.util.Map;
8 import java.util.Map.Entry;
9 import java.util.concurrent.ExecutorService;
10 import java.util.concurrent.Executors;
11 import java.util.concurrent.ThreadFactory;
12
13 import org.opendaylight.controller.md.statistics.manager.StatPermCollector;
14 import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
15 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
16 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
17 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId;
18 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
19 import org.slf4j.Logger;
20 import org.slf4j.LoggerFactory;
21
22 import com.google.common.base.Preconditions;
23 import com.google.common.util.concurrent.ThreadFactoryBuilder;
24
25 /**
26  * statistics-manager
27  * org.opendaylight.controller.md.statistics.manager.impl
28  *
29  * StatPermCollectorImpl
30  * Thread base statistic collector. Class holds internal map for all registered
31  * (means connected) nodes with List of Switch capabilities;
32  * Statistics collecting process get cross whole Network Device by device
33  * and statistic by statistic (follow Switch capabilities to prevent unnecessary
34  * ask) Next statistic start collecting by notification or by timeout.
35  *
36  * @author @author avishnoi@in.ibm.com <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
37  *
38  */
39 public class StatPermCollectorImpl implements StatPermCollector {
40
41     private final static Logger LOG = LoggerFactory.getLogger(StatPermCollectorImpl.class);
42
43     private final static long STAT_COLLECT_TIME_OUT = 30000L;
44
45     private final ExecutorService statNetCollectorServ;
46     private final StatisticsManager manager;
47
48     private final int maxNodeForCollector;
49     private final long minReqNetInterval;
50     private final String name;
51
52     private final Object statCollectorLock = new Object();
53     private final Object statNodeHolderLock = new Object();
54
55     private Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNodeHolder =
56             Collections.<InstanceIdentifier<Node>, StatNodeInfoHolder> emptyMap();
57
58     private volatile boolean wakeMe = false;
59     private volatile boolean finishing = false;
60
61     public StatPermCollectorImpl(final StatisticsManager manager, final long minReqNetInterv, final int nr,
62             final int maxNodeForCollectors) {
63         this.manager = Preconditions.checkNotNull(manager, "StatisticsManager can not be null!");
64         name = "odl-stat-collector-" + nr;
65         minReqNetInterval = minReqNetInterv;
66         final ThreadFactory threadFact = new ThreadFactoryBuilder()
67             .setNameFormat(name + "-thread-%d").build();
68         statNetCollectorServ = Executors.newSingleThreadExecutor(threadFact);
69         maxNodeForCollector = maxNodeForCollectors;
70         LOG.trace("StatCollector {} start successfull!", name);
71     }
72
73     /**
74      * finish collecting statistics
75      */
76     @Override
77     public void close() {
78         statNodeHolder = Collections.<InstanceIdentifier<Node>, StatNodeInfoHolder> emptyMap();
79         finishing = true;
80         collectNextStatistics();
81         statNetCollectorServ.shutdown();
82     }
83
84     @Override
85     public boolean hasActiveNodes() {
86         return ( ! statNodeHolder.isEmpty());
87     }
88
89     @Override
90     public boolean isProvidedFlowNodeActive(
91             final InstanceIdentifier<Node> flowNode) {
92         return statNodeHolder.containsKey(flowNode);
93     }
94
95     @Override
96     public boolean connectedNodeRegistration(final InstanceIdentifier<Node> ident,
97             final List<StatCapabTypes> statTypes, final Short nrOfSwitchTables) {
98         if (isNodeIdentValidForUse(ident)) {
99             if ( ! statNodeHolder.containsKey(ident)) {
100                 synchronized (statNodeHolderLock) {
101                     final boolean startStatCollecting = statNodeHolder.size() == 0;
102                     if ( ! statNodeHolder.containsKey(ident)) {
103                         if (statNodeHolder.size() >= maxNodeForCollector) {
104                             return false;
105                         }
106                         final Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNode =
107                                 new HashMap<>(statNodeHolder);
108                         final NodeRef nodeRef = new NodeRef(ident);
109                         final StatNodeInfoHolder nodeInfoHolder = new StatNodeInfoHolder(nodeRef,
110                                 statTypes, nrOfSwitchTables);
111                         statNode.put(ident, nodeInfoHolder);
112                         statNodeHolder = Collections.unmodifiableMap(statNode);
113                     }
114                     if (startStatCollecting) {
115                         finishing = false;
116                         statNetCollectorServ.execute(this);
117                     }
118                 }
119             }
120         }
121         return true;
122     }
123
124     @Override
125     public boolean disconnectedNodeUnregistration(final InstanceIdentifier<Node> ident) {
126         if (isNodeIdentValidForUse(ident)) {
127             if (statNodeHolder.containsKey(ident)) {
128                 synchronized (statNodeHolderLock) {
129                     if (statNodeHolder.containsKey(ident)) {
130                         final Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNode =
131                                 new HashMap<>(statNodeHolder);
132                         statNode.remove(ident);
133                         statNodeHolder = Collections.unmodifiableMap(statNode);
134                     }
135                     if (statNodeHolder.isEmpty()) {
136                         finishing = true;
137                         collectNextStatistics();
138                         statNetCollectorServ.shutdown();
139                     }
140                     return true;
141                 }
142             }
143         }
144         return false;
145     }
146
147     @Override
148     public boolean registerAdditionalNodeFeature(final InstanceIdentifier<Node> ident,
149             final StatCapabTypes statCapab) {
150         if (isNodeIdentValidForUse(ident)) {
151             if ( ! statNodeHolder.containsKey(ident)) {
152                 return false;
153             }
154             final StatNodeInfoHolder statNode = statNodeHolder.get(ident);
155             if ( ! statNode.getStatMarkers().contains(statCapab)) {
156                 synchronized (statNodeHolderLock) {
157                     if ( ! statNode.getStatMarkers().contains(statCapab)) {
158                         final List<StatCapabTypes> statCapabForEdit = new ArrayList<>(statNode.getStatMarkers());
159                         statCapabForEdit.add(statCapab);
160                         final StatNodeInfoHolder nodeInfoHolder = new StatNodeInfoHolder(statNode.getNodeRef(),
161                                 Collections.unmodifiableList(statCapabForEdit), statNode.getMaxTables());
162
163                         final Map<InstanceIdentifier<Node>, StatNodeInfoHolder> statNodes =
164                                 new HashMap<>(statNodeHolder);
165                         statNodes.put(ident, nodeInfoHolder);
166                         statNodeHolder = Collections.unmodifiableMap(statNodes);
167                     }
168                 }
169             }
170         }
171         return true;
172     }
173
174     @Override
175     public void collectNextStatistics() {
176         if (wakeMe) {
177             synchronized (statCollectorLock) {
178                 if (wakeMe) {
179                     LOG.trace("STAT-COLLECTOR is notified to conntinue");
180                     statCollectorLock.notify();
181                 }
182             }
183         }
184     }
185
186     @Override
187     public void run() {
188         try {
189             Thread.sleep(5000);
190         }
191         catch (final InterruptedException e1) {
192             // NOOP
193         }
194         LOG.debug("StatCollector {} Start collecting!", name);
195          /* Neverending cyle - wait for finishing */
196          while ( ! finishing) {
197             boolean collecting = false;
198             final long startTime = System.currentTimeMillis();
199
200             if ( ! statNodeHolder.isEmpty()) {
201                 collecting = true;
202                 collectStatCrossNetwork();
203                 collecting = false;
204             }
205
206             if ( ! collecting) {
207                 final long statFinalTime = System.currentTimeMillis() - startTime;
208                 LOG.debug("STAT-MANAGER {}: last all NET statistics collection cost {} ms", name, statFinalTime);
209                 if (statFinalTime < minReqNetInterval) {
210                     LOG.trace("statCollector is about to make a collecting sleep");
211                     synchronized (statCollectorLock) {
212                         wakeMe = true;
213                         try {
214                             final long waitTime = minReqNetInterval - statFinalTime;
215                             statCollectorLock.wait(waitTime);
216                             LOG.trace("STAT-MANAGER : statCollector {} is waking up from a collecting sleep for {} ms", name, waitTime);
217                         } catch (final InterruptedException e) {
218                             LOG.warn("statCollector has been interrupted during collecting sleep", e);
219                         } finally {
220                             wakeMe = false;
221                         }
222                     }
223                 }
224             }
225         }
226     }
227
228     private void waitingForNotification() {
229         synchronized (statCollectorLock) {
230             wakeMe = true;
231             try {
232                 statCollectorLock.wait(STAT_COLLECT_TIME_OUT);
233                 LOG.trace("statCollector is waking up from a wait stat Response sleep");
234             } catch (final InterruptedException e) {
235                 LOG.warn("statCollector has been interrupted waiting stat Response sleep", e);
236             } finally {
237                 wakeMe = false;
238             }
239         }
240     }
241
242
243     private void collectStatCrossNetwork() {
244         for (final Entry<InstanceIdentifier<Node>, StatNodeInfoHolder> nodeEntity : statNodeHolder.entrySet()) {
245             final List<StatCapabTypes> listNeededStat = nodeEntity.getValue().getStatMarkers();
246             final NodeRef actualNodeRef = nodeEntity.getValue().getNodeRef();
247             final Short maxTables = nodeEntity.getValue().getMaxTables();
248             for (final StatCapabTypes statMarker : listNeededStat) {
249                 if ( ! isProvidedFlowNodeActive(nodeEntity.getKey())) {
250                     break;
251                 }
252                 switch (statMarker) {
253                 case PORT_STATS:
254                     LOG.trace("STAT-MANAGER-collecting PORT-STATS for NodeRef {}", actualNodeRef);
255                     manager.getRpcMsgManager().getAllPortsStat(actualNodeRef);
256                     waitingForNotification();
257                     break;
258                 case QUEUE_STATS:
259                     LOG.trace("STAT-MANAGER-collecting QUEUE-STATS for NodeRef {}", actualNodeRef);
260                     manager.getRpcMsgManager().getAllQueueStat(actualNodeRef);
261                     waitingForNotification();
262                     break;
263                 case TABLE_STATS:
264                     LOG.trace("STAT-MANAGER-collecting TABLE-STATS for NodeRef {}", actualNodeRef);
265                     manager.getRpcMsgManager().getAllTablesStat(actualNodeRef);
266                     waitingForNotification();
267                     break;
268                 case GROUP_STATS:
269                     LOG.trace("STAT-MANAGER-collecting GROUP-STATS for NodeRef {}", actualNodeRef);
270                     manager.getRpcMsgManager().getAllGroupsConfStats(actualNodeRef);
271                     waitingForNotification();
272                     manager.getRpcMsgManager().getAllGroupsStat(actualNodeRef);
273                     waitingForNotification();
274                     break;
275                 case METER_STATS:
276                     LOG.trace("STAT-MANAGER-collecting METER-STATS for NodeRef {}", actualNodeRef);
277                     manager.getRpcMsgManager().getAllMeterConfigStat(actualNodeRef);
278                     waitingForNotification();
279                     manager.getRpcMsgManager().getAllMetersStat(actualNodeRef);
280                     waitingForNotification();
281                     break;
282                 case FLOW_STATS:
283                     LOG.trace("STAT-MANAGER-collecting FLOW-STATS-ALL_FLOWS for NodeRef {}", actualNodeRef);
284                     manager.getRpcMsgManager().getAllFlowsStat(actualNodeRef);
285                     waitingForNotification();
286                     LOG.trace("STAT-MANAGER-collecting FLOW-AGGREGATE-STATS for NodeRef {}", actualNodeRef);
287                     for (short i = 0; i < maxTables; i++) {
288                         final TableId tableId = new TableId(i);
289                         manager.getRpcMsgManager().getAggregateFlowStat(actualNodeRef, tableId);
290                     }
291                     break;
292                 default:
293                     /* Exception for programmers in implementation cycle */
294                     throw new IllegalStateException("Not implemented ASK for " + statMarker);
295                 }
296             }
297         }
298     }
299
300     private class StatNodeInfoHolder {
301         private final NodeRef nodeRef;
302         private final List<StatCapabTypes> statMarkers;
303         private final Short maxTables;
304
305         public StatNodeInfoHolder(final NodeRef nodeRef,
306                 final List<StatCapabTypes> statMarkers, final Short maxTables) {
307             this.nodeRef = nodeRef;
308             this.maxTables = maxTables;
309             this.statMarkers = statMarkers;
310         }
311
312         public final NodeRef getNodeRef() {
313             return nodeRef;
314         }
315
316         public final List<StatCapabTypes> getStatMarkers() {
317             return statMarkers;
318         }
319
320         public final Short getMaxTables() {
321             return maxTables;
322         }
323     }
324
325     private boolean isNodeIdentValidForUse(final InstanceIdentifier<Node> ident) {
326         if (ident == null) {
327             LOG.warn("FlowCapableNode InstanceIdentifier {} can not be null!");
328             return false;
329         }
330         if (ident.isWildcarded()) {
331             LOG.warn("FlowCapableNode InstanceIdentifier {} can not be wildcarded!", ident);
332             return false;
333         }
334         return true;
335     }
336 }
337

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.