2 * Copyright IBM Corporation, 2013. 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.controller.md.statistics.manager;
10 import java.util.ArrayList;
11 import java.util.List;
12 import java.util.concurrent.ConcurrentHashMap;
13 import java.util.concurrent.ConcurrentMap;
14 import java.util.concurrent.ExecutionException;
15 import java.util.concurrent.Future;
17 import org.eclipse.xtext.xbase.lib.Exceptions;
18 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
19 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
20 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesOutput;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsOutput;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsInputBuilder;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsOutput;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionInputBuilder;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionOutput;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsService;
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.meter.statistics.rev131111.GetAllMeterConfigStatisticsInputBuilder;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsOutput;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsInputBuilder;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsOutput;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsService;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllPortsStatisticsInputBuilder;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllPortsStatisticsOutput;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsInputBuilder;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsOutput;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsService;
52 import org.opendaylight.yangtools.concepts.Registration;
53 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
54 import org.opendaylight.yangtools.yang.binding.NotificationListener;
55 import org.opendaylight.yangtools.yang.common.RpcResult;
56 import org.slf4j.Logger;
57 import org.slf4j.LoggerFactory;
59 public class StatisticsProvider implements AutoCloseable {
61 public final static Logger spLogger = LoggerFactory.getLogger(StatisticsProvider.class);
63 private DataProviderService dps;
65 private NotificationProviderService nps;
67 private OpendaylightGroupStatisticsService groupStatsService;
69 private OpendaylightMeterStatisticsService meterStatsService;
71 private OpendaylightFlowStatisticsService flowStatsService;
73 private OpendaylightPortStatisticsService portStatsService;
75 private OpendaylightFlowTableStatisticsService flowTableStatsService;
77 private OpendaylightQueueStatisticsService queueStatsService;
79 private final MultipartMessageManager multipartMessageManager = new MultipartMessageManager();
81 private Thread statisticsRequesterThread;
83 private final InstanceIdentifier<Nodes> nodesIdentifier = InstanceIdentifier.builder(Nodes.class).toInstance();
85 private final int STATS_THREAD_EXECUTION_TIME= 50000;
86 //Local caching of stats
88 private final ConcurrentMap<NodeId,NodeStatistics> statisticsCache =
89 new ConcurrentHashMap<NodeId,NodeStatistics>();
91 public DataProviderService getDataService() {
95 public void setDataService(final DataProviderService dataService) {
96 this.dps = dataService;
99 public NotificationProviderService getNotificationService() {
103 public void setNotificationService(final NotificationProviderService notificationService) {
104 this.nps = notificationService;
107 public MultipartMessageManager getMultipartMessageManager() {
108 return multipartMessageManager;
111 private final StatisticsUpdateCommiter updateCommiter = new StatisticsUpdateCommiter(StatisticsProvider.this);
113 private Registration<NotificationListener> listenerRegistration;
115 public void start() {
117 NotificationProviderService nps = this.getNotificationService();
118 Registration<NotificationListener> registerNotificationListener = nps.registerNotificationListener(this.updateCommiter);
119 this.listenerRegistration = registerNotificationListener;
121 // Get Group/Meter statistics service instance
122 groupStatsService = StatisticsManagerActivator.getProviderContext().
123 getRpcService(OpendaylightGroupStatisticsService.class);
125 meterStatsService = StatisticsManagerActivator.getProviderContext().
126 getRpcService(OpendaylightMeterStatisticsService.class);
128 flowStatsService = StatisticsManagerActivator.getProviderContext().
129 getRpcService(OpendaylightFlowStatisticsService.class);
131 portStatsService = StatisticsManagerActivator.getProviderContext().
132 getRpcService(OpendaylightPortStatisticsService.class);
134 flowTableStatsService = StatisticsManagerActivator.getProviderContext().
135 getRpcService(OpendaylightFlowTableStatisticsService.class);
137 queueStatsService = StatisticsManagerActivator.getProviderContext().
138 getRpcService(OpendaylightQueueStatisticsService.class);
140 statisticsRequesterThread = new Thread( new Runnable(){
146 statsRequestSender();
148 Thread.sleep(STATS_THREAD_EXECUTION_TIME);
149 }catch (Exception e){
150 spLogger.error("Exception occurred while sending stats request : {}",e);
156 spLogger.debug("Statistics requester thread started with timer interval : {}",STATS_THREAD_EXECUTION_TIME);
158 statisticsRequesterThread.start();
160 spLogger.info("Statistics Provider started.");
163 protected DataModificationTransaction startChange() {
165 DataProviderService dps = this.getDataService();
166 return dps.beginTransaction();
169 private void statsRequestSender(){
171 List<Node> targetNodes = getAllConnectedNodes();
173 if(targetNodes == null)
177 for (Node targetNode : targetNodes){
179 InstanceIdentifier<Node> targetInstanceId = InstanceIdentifier.builder(Nodes.class).child(Node.class,targetNode.getKey()).toInstance();
180 NodeRef targetNodeRef = new NodeRef(targetInstanceId);
184 sendAggregateFlowsStatsFromAllTablesRequest(targetNode.getKey());
186 sendAllFlowsStatsFromAllTablesRequest(targetNodeRef);
188 sendAllPortStatisticsRequest(targetNodeRef);
190 sendAllFlowTablesStatisticsRequest(targetNodeRef);
192 sendAllQueueStatsFromAllNodeConnector (targetNodeRef);
195 spLogger.error("Exception occured while sending statistics requests : {}",e);
198 if(targetNode.getAugmentation(FlowCapableNode.class) != null){
200 spLogger.info("Send request for stats collection to node : {})",targetNode.getId());
203 sendAllGroupStatisticsRequest(targetNodeRef);
205 sendAllMeterStatisticsRequest(targetNodeRef);
207 sendGroupDescriptionRequest(targetNodeRef);
209 sendMeterConfigStatisticsRequest(targetNodeRef);
212 spLogger.error("Exception occured while sending statistics requests : {}", e);
218 private void sendAllFlowTablesStatisticsRequest(NodeRef targetNodeRef) {
219 final GetFlowTablesStatisticsInputBuilder input =
220 new GetFlowTablesStatisticsInputBuilder();
222 input.setNode(targetNodeRef);
224 @SuppressWarnings("unused")
225 Future<RpcResult<GetFlowTablesStatisticsOutput>> response =
226 flowTableStatsService.getFlowTablesStatistics(input.build());
229 private void sendAllFlowsStatsFromAllTablesRequest(NodeRef targetNode){
230 final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input =
231 new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
233 input.setNode(targetNode);
235 @SuppressWarnings("unused")
236 Future<RpcResult<GetAllFlowsStatisticsFromAllFlowTablesOutput>> response =
237 flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build());
241 private void sendAggregateFlowsStatsFromAllTablesRequest(NodeKey targetNodeKey) throws InterruptedException, ExecutionException{
243 List<Short> tablesId = getTablesFromNode(targetNodeKey);
245 if(tablesId.size() != 0){
246 for(Short id : tablesId){
248 spLogger.info("Send aggregate stats request for flow table {} to node {}",id,targetNodeKey);
249 GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
250 new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
252 input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
253 input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(id));
254 Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
255 flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());
257 multipartMessageManager.setTxIdAndTableIdMapEntry(response.get().getResult().getTransactionId(), id);
261 //Note: Just for testing, because i am not able to fetch table list from datastore
262 // Bug-225 is raised for investigation.
264 // spLogger.info("Send aggregate stats request for flow table {} to node {}",1,targetNodeKey);
265 // GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
266 // new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
268 // input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
269 // input.setTableId(new TableId((short)1));
270 // Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
271 // flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());`
273 // multipartMessageManager.setTxIdAndTableIdMapEntry(response.get().getResult().getTransactionId(), (short)1);
276 private void sendAllPortStatisticsRequest(NodeRef targetNode){
278 final GetAllPortsStatisticsInputBuilder input = new GetAllPortsStatisticsInputBuilder();
280 input.setNode(targetNode);
282 @SuppressWarnings("unused")
283 Future<RpcResult<GetAllPortsStatisticsOutput>> response =
284 portStatsService.getAllPortsStatistics(input.build());
287 private void sendAllGroupStatisticsRequest(NodeRef targetNode){
289 final GetAllGroupStatisticsInputBuilder input = new GetAllGroupStatisticsInputBuilder();
291 input.setNode(targetNode);
293 @SuppressWarnings("unused")
294 Future<RpcResult<GetAllGroupStatisticsOutput>> response =
295 groupStatsService.getAllGroupStatistics(input.build());
298 private void sendGroupDescriptionRequest(NodeRef targetNode){
299 final GetGroupDescriptionInputBuilder input = new GetGroupDescriptionInputBuilder();
301 input.setNode(targetNode);
303 @SuppressWarnings("unused")
304 Future<RpcResult<GetGroupDescriptionOutput>> response =
305 groupStatsService.getGroupDescription(input.build());
308 private void sendAllMeterStatisticsRequest(NodeRef targetNode){
310 GetAllMeterStatisticsInputBuilder input = new GetAllMeterStatisticsInputBuilder();
312 input.setNode(targetNode);
314 @SuppressWarnings("unused")
315 Future<RpcResult<GetAllMeterStatisticsOutput>> response =
316 meterStatsService.getAllMeterStatistics(input.build());
319 private void sendMeterConfigStatisticsRequest(NodeRef targetNode){
321 GetAllMeterConfigStatisticsInputBuilder input = new GetAllMeterConfigStatisticsInputBuilder();
323 input.setNode(targetNode);
325 @SuppressWarnings("unused")
326 Future<RpcResult<GetAllMeterConfigStatisticsOutput>> response =
327 meterStatsService.getAllMeterConfigStatistics(input.build());
330 private void sendAllQueueStatsFromAllNodeConnector(NodeRef targetNode) {
331 GetAllQueuesStatisticsFromAllPortsInputBuilder input = new GetAllQueuesStatisticsFromAllPortsInputBuilder();
333 input.setNode(targetNode);
335 @SuppressWarnings("unused")
336 Future<RpcResult<GetAllQueuesStatisticsFromAllPortsOutput>> response =
337 queueStatsService.getAllQueuesStatisticsFromAllPorts(input.build());
340 public ConcurrentMap<NodeId, NodeStatistics> getStatisticsCache() {
341 return statisticsCache;
344 private List<Node> getAllConnectedNodes(){
346 Nodes nodes = (Nodes) dps.readOperationalData(nodesIdentifier);
350 spLogger.info("Number of connected nodes : {}",nodes.getNode().size());
351 return nodes.getNode();
354 private List<Short> getTablesFromNode(NodeKey nodeKey){
355 InstanceIdentifier<FlowCapableNode> nodesIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class,nodeKey).augmentation(FlowCapableNode.class).toInstance();
357 FlowCapableNode node = (FlowCapableNode)dps.readConfigurationData(nodesIdentifier);
358 List<Short> tablesId = new ArrayList<Short>();
359 if(node != null && node.getTable()!=null){
360 spLogger.info("Number of tables {} supported by node {}",node.getTable().size(),nodeKey);
361 for(Table table: node.getTable()){
362 tablesId.add(table.getId());
368 @SuppressWarnings("deprecation")
373 spLogger.info("Statistics Provider stopped.");
374 if (this.listenerRegistration != null) {
376 this.listenerRegistration.close();
378 this.statisticsRequesterThread.destroy();
381 } catch (Throwable e) {
382 throw Exceptions.sneakyThrow(e);