2 * Copyright (c) 2016 Ericsson India Global Services Pvt Ltd. 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
8 package org.opendaylight.genius.interfacemanager.pmcounters;
10 import com.google.common.util.concurrent.ThreadFactoryBuilder;
11 import java.math.BigInteger;
12 import java.util.ArrayList;
13 import java.util.HashMap;
14 import java.util.List;
16 import java.util.concurrent.ConcurrentHashMap;
17 import java.util.concurrent.Executors;
18 import java.util.concurrent.ScheduledExecutorService;
19 import java.util.concurrent.ScheduledFuture;
20 import java.util.concurrent.ThreadFactory;
21 import java.util.concurrent.TimeUnit;
22 import javax.inject.Inject;
23 import javax.inject.Singleton;
24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
25 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
26 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
28 import org.opendaylight.genius.mdsalutil.AbstractDataChangeListener;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsUpdate;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInput;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsListener;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInput;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInputBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.NodeConnectorStatisticsUpdate;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsListener;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
48 import org.slf4j.Logger;
49 import org.slf4j.LoggerFactory;
52 public class NodeConnectorStatsImpl extends AbstractDataChangeListener<Node>{
54 private static final Logger logger = LoggerFactory.getLogger(NodeConnectorStatsImpl.class);
55 private static final String STATS_POLL_FLAG = "interfacemgr.pmcounters.poll";
56 private static final int THREAD_POOL_SIZE = 4;
57 private static final int NO_DELAY = 0;
58 public static final PMAgentForNodeConnectorCounters pmagent = new PMAgentForNodeConnectorCounters();
59 private PortRpcStatisticsListener portStatsListener = new PortRpcStatisticsListener();
60 private FlowRpcStatisticsListener flowTableStatsListener = new FlowRpcStatisticsListener();
61 private List<BigInteger> nodes = new ArrayList<>();
62 Map<String, Map<String, String>> nodeAndNcIdOFPortDurationMap = new ConcurrentHashMap<>();
63 Map<String, Map<String, String>> nodeAndNcIdOFPortReceiveDropMap = new ConcurrentHashMap<>();
64 Map<String, Map<String, String>> nodeAndNcIdOFPortReceiveError = new ConcurrentHashMap<>();
65 Map<String, Map<String, String>> nodeAndNcIdPacketSentMap = new ConcurrentHashMap<>();
66 Map<String, Map<String, String>> nodeAndNcIdPacketReceiveMap = new ConcurrentHashMap<>();
67 Map<String, Map<String, String>> nodeAndNcIdBytesSentMap = new ConcurrentHashMap<>();
68 Map<String, Map<String, String>> nodeAndNcIdBytesReceiveMap = new ConcurrentHashMap<>();
69 Map<String, Map<String, String>> nodeAndEntriesPerOFTableMap = new ConcurrentHashMap<>();
70 private ScheduledFuture<?> scheduledResult;
71 private final OpendaylightPortStatisticsService statPortService;
72 private ScheduledExecutorService portStatExecutorService;
73 private final OpendaylightFlowTableStatisticsService opendaylightFlowTableStatisticsService;
76 public NodeConnectorStatsImpl(DataBroker dataBroker, NotificationService notificationService,
77 final OpendaylightPortStatisticsService opendaylightPortStatisticsService,
78 final OpendaylightFlowTableStatisticsService opendaylightFlowTableStatisticsService) {
80 this.statPortService = opendaylightPortStatisticsService;
81 this.opendaylightFlowTableStatisticsService = opendaylightFlowTableStatisticsService;
82 registerListener(dataBroker);
83 portStatExecutorService = Executors.newScheduledThreadPool(THREAD_POOL_SIZE, getThreadFactory("Port Stats Request Task"));
84 notificationService.registerNotificationListener(portStatsListener);
85 notificationService.registerNotificationListener(flowTableStatsListener);
86 pmagent.registerMbean();
89 private void registerListener(final DataBroker db) {
91 db.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL,
92 getWildCardPath(), NodeConnectorStatsImpl.this, AsyncDataBroker.DataChangeScope.SUBTREE);
93 } catch (final Exception e) {
94 logger.error("NodeConnectorStatsImpl: DataChange listener registration fail!", e);
95 throw new IllegalStateException("NodeConnectorStatsImpl: registration Listener failed.", e);
99 private InstanceIdentifier<Node> getWildCardPath() {
100 return InstanceIdentifier.create(Nodes.class).child(Node.class);
104 * PortStat request task is started when first DPN gets connected
106 private void schedulePortStatRequestTask() {
107 if (!Boolean.getBoolean(STATS_POLL_FLAG)) {
108 logger.info("Port statistics is turned off");
111 logger.info("Scheduling port statistics request");
112 PortStatRequestTask portStatRequestTask = new PortStatRequestTask();
113 scheduledResult = portStatExecutorService.scheduleAtFixedRate(portStatRequestTask, NO_DELAY, 10000, TimeUnit.MILLISECONDS);
117 * PortStat request task is stopped when last DPN is removed.
119 private void stopPortStatRequestTask() {
120 if(scheduledResult != null) {
121 logger.info("Stopping port statistics request");
122 scheduledResult.cancel(true);
127 * This task queries for node connector statistics as well as flowtables statistics every 10 secs.
128 * Minimum period which can be configured for PMJob is 10 secs.
130 private class PortStatRequestTask implements Runnable {
134 if(logger.isTraceEnabled()) {
135 logger.trace("Requesting port stats - {}");
137 for (BigInteger node : nodes) {
138 logger.trace("Requesting AllNodeConnectorStatistics for node - {}", node);
139 statPortService.getAllNodeConnectorsStatistics(buildGetAllNodeConnectorStatistics(node));
140 opendaylightFlowTableStatisticsService.getFlowTablesStatistics(buildGetFlowTablesStatistics(node));
144 private GetAllNodeConnectorsStatisticsInput buildGetAllNodeConnectorStatistics(BigInteger dpId) {
145 return new GetAllNodeConnectorsStatisticsInputBuilder()
147 new NodeRef(InstanceIdentifier.builder(Nodes.class)
148 .child(Node.class, new NodeKey(new NodeId("openflow:" + dpId.toString()))).build())).build();
151 private GetFlowTablesStatisticsInput buildGetFlowTablesStatistics(BigInteger dpId) {
152 return new GetFlowTablesStatisticsInputBuilder()
154 new NodeRef(InstanceIdentifier.builder(Nodes.class)
155 .child(Node.class, new NodeKey(new NodeId("openflow:" + dpId.toString()))).build())).build();
160 private ThreadFactory getThreadFactory(String threadNameFormat) {
161 ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
162 builder.setNameFormat(threadNameFormat);
163 builder.setUncaughtExceptionHandler(
164 (t, e) -> logger.error("Received Uncaught Exception event in Thread: {}", t.getName(), e));
165 return builder.build();
169 * PortRpcStatisticsListener listens for the NodeConnectorStatisticsUpdate and then update the corresponding counter map
171 class PortRpcStatisticsListener implements OpendaylightPortStatisticsListener {
174 public void onNodeConnectorStatisticsUpdate(NodeConnectorStatisticsUpdate ncStats) {
175 Map<String, String> ncIdOFPortDurationMap = new HashMap<>();
176 Map<String, String> ncIdOFPortReceiveDropMap = new HashMap<>();
177 Map<String, String> ncIdOFPortReceiveError = new HashMap<>();
178 Map<String, String> ncIdPacketSentMap = new HashMap<>();
179 Map<String, String> ncIdPacketReceiveMap = new HashMap<>();
180 Map<String, String> ncIdBytesSentMap = new HashMap<>();
181 Map<String, String> ncIdBytesReceiveMap = new HashMap<>();
182 List<NodeConnectorStatisticsAndPortNumberMap> ncStatsAndPortMapList = ncStats.getNodeConnectorStatisticsAndPortNumberMap();
183 NodeId nodeId = ncStats.getId();
184 String node = nodeId.getValue().split(":")[1];
185 for (NodeConnectorStatisticsAndPortNumberMap ncStatsAndPortMap : ncStatsAndPortMapList) {
186 NodeConnectorId nodeConnector = ncStatsAndPortMap.getNodeConnectorId();
187 String port = nodeConnector.getValue().split(":")[2];
188 String nodePortStr = "dpnId_" + node + "_portNum_" + port;
189 ncIdOFPortDurationMap.put("OFPortDuration:" + nodePortStr + "_OFPortDuration", ncStatsAndPortMap.getDuration().getSecond().getValue().toString());
190 ncIdOFPortReceiveDropMap.put("PacketsPerOFPortReceiveDrop:" + nodePortStr + "_PacketsPerOFPortReceiveDrop", ncStatsAndPortMap.getReceiveDrops().toString());
191 ncIdOFPortReceiveError.put("PacketsPerOFPortReceiveError:" + nodePortStr + "_PacketsPerOFPortReceiveError", ncStatsAndPortMap.getReceiveErrors().toString());
192 ncIdPacketSentMap.put("PacketsPerOFPortSent:" + nodePortStr + "_PacketsPerOFPortSent", ncStatsAndPortMap.getPackets().getTransmitted().toString());
193 ncIdPacketReceiveMap.put("PacketsPerOFPortReceive:" + nodePortStr + "_PacketsPerOFPortReceive", ncStatsAndPortMap.getPackets().getReceived().toString());
194 ncIdBytesSentMap.put("BytesPerOFPortSent:" + nodePortStr + "_BytesPerOFPortSent", ncStatsAndPortMap.getBytes().getTransmitted().toString());
195 ncIdBytesReceiveMap.put("BytesPerOFPortReceive:" + nodePortStr + "_BytesPerOFPortReceive", ncStatsAndPortMap.getBytes().getReceived().toString());
197 logger.trace("Port Stats {}", ncStatsAndPortMapList);
198 //Storing allNodeConnectorStats(like ncIdOFPortDurationMap) in a map with key as node for easy removal and addition of allNodeConnectorStats.
199 nodeAndNcIdOFPortDurationMap.put(node, ncIdOFPortDurationMap);
200 nodeAndNcIdOFPortReceiveDropMap.put(node, ncIdOFPortReceiveDropMap);
201 nodeAndNcIdOFPortReceiveError.put(node, ncIdOFPortReceiveError);
202 nodeAndNcIdPacketSentMap.put(node, ncIdPacketSentMap);
203 nodeAndNcIdPacketReceiveMap.put(node, ncIdPacketReceiveMap);
204 nodeAndNcIdBytesSentMap.put(node, ncIdBytesSentMap);
205 nodeAndNcIdBytesReceiveMap.put(node, ncIdBytesReceiveMap);
206 //Combining the stats of all nodeconnectors in all nodes. This Map will be stored under MBean which will be queried as regular intervals.
207 ncIdOFPortDurationMap = combineAllNodesStats(nodeAndNcIdOFPortDurationMap);
208 ncIdOFPortReceiveDropMap = combineAllNodesStats(nodeAndNcIdOFPortReceiveDropMap);
209 ncIdOFPortReceiveError = combineAllNodesStats(nodeAndNcIdOFPortReceiveError);
210 ncIdPacketSentMap = combineAllNodesStats(nodeAndNcIdPacketSentMap);
211 ncIdPacketReceiveMap = combineAllNodesStats(nodeAndNcIdPacketReceiveMap);
212 ncIdBytesSentMap = combineAllNodesStats(nodeAndNcIdBytesSentMap);
213 ncIdBytesReceiveMap = combineAllNodesStats(nodeAndNcIdBytesReceiveMap);
214 pmagent.connectToPMAgent(ncIdOFPortDurationMap, ncIdOFPortReceiveDropMap, ncIdOFPortReceiveError, ncIdPacketSentMap, ncIdPacketReceiveMap, ncIdBytesSentMap, ncIdBytesReceiveMap);
218 * Input allNodesStats contains statistics of all nodeConnectors of all nodes. Key is the node and values contains another map with key as node connector and value as statresult.
219 * Output will be a map with key as nodeconnector and value as the statresult. The key contains nodeconnectors of all the nodes.
223 private Map<String, String> combineAllNodesStats(Map<String, Map<String, String>> allNodesStats) {
224 Map<String, String> allNcsStatsMap = new HashMap<>();
225 for (Map.Entry<String, Map<String, String>> entry : allNodesStats.entrySet()) {
226 Map<String, String> ncStatsMap = entry.getValue();
227 for (Map.Entry<String, String> statResult : ncStatsMap.entrySet()) {
228 allNcsStatsMap.put(statResult.getKey(), statResult.getValue());
231 return allNcsStatsMap;
235 * FlowRpcStatisticsListener listens for the FlowTableStatisticsUpdate and then update the corresponding counter map
237 class FlowRpcStatisticsListener implements OpendaylightFlowTableStatisticsListener {
240 public void onFlowTableStatisticsUpdate(FlowTableStatisticsUpdate flowTableStats) {
241 String node = flowTableStats.getId().getValue().split(":")[1];
242 Map<String, String> entriesPerOFTableMap = new HashMap<>();
243 List<FlowTableAndStatisticsMap> flowTableAndStatisticsMapList = flowTableStats.getFlowTableAndStatisticsMap();
244 for (FlowTableAndStatisticsMap flowTableAndStatisticsMap : flowTableAndStatisticsMapList) {
245 String nodeTableStr = "dpnId_" + node + "_table_" + flowTableAndStatisticsMap.getTableId().getValue().toString();
246 entriesPerOFTableMap.put("EntriesPerOFTable:" + nodeTableStr + "_EntriesPerOFTable", flowTableAndStatisticsMap.getActiveFlows().getValue().toString());
248 nodeAndEntriesPerOFTableMap.put(node, entriesPerOFTableMap);
249 entriesPerOFTableMap = combineAllNodesStats(nodeAndEntriesPerOFTableMap);
250 pmagent.connectToPMAgentAndInvokeEntriesPerOFTable(entriesPerOFTableMap);
256 protected void remove(InstanceIdentifier<Node> identifier, Node node) {
257 NodeId nodeId = node.getId();
258 String nodeVal = nodeId.getValue().split(":")[1];
259 BigInteger dpId = new BigInteger(nodeVal);
260 if (nodes.contains(dpId)) {
262 nodeAndNcIdOFPortDurationMap.remove(nodeVal);
263 nodeAndNcIdOFPortReceiveDropMap.remove(nodeVal);
264 nodeAndNcIdOFPortReceiveError.remove(nodeVal);
265 nodeAndNcIdPacketSentMap.remove(nodeVal);
266 nodeAndNcIdPacketReceiveMap.remove(nodeVal);
267 nodeAndNcIdBytesSentMap.remove(nodeVal);
268 nodeAndNcIdBytesReceiveMap.remove(nodeVal);
269 nodeAndEntriesPerOFTableMap.remove(nodeVal);
271 if (nodes.isEmpty()) {
272 stopPortStatRequestTask();
277 protected void update(InstanceIdentifier<Node> identifier, Node original,
279 // TODO Auto-generated method stub
283 protected void add(InstanceIdentifier<Node> identifier, Node node) {
284 NodeId nodeId = node.getId();
285 BigInteger dpId = new BigInteger(nodeId.getValue().split(":")[1]);
286 if (nodes.contains(dpId)) {
290 if (nodes.size() == 1) {
291 schedulePortStatRequestTask();