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.md.statistics.manager.MultipartMessageManager.StatsRequestType;
19 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
20 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
21 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesOutput;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsOutput;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsInputBuilder;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsOutput;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionInputBuilder;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionOutput;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsService;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsInputBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsOutput;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsInputBuilder;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsOutput;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsService;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllPortsStatisticsInputBuilder;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllPortsStatisticsOutput;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsInputBuilder;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsOutput;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsService;
53 import org.opendaylight.yangtools.concepts.Registration;
54 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
55 import org.opendaylight.yangtools.yang.binding.NotificationListener;
56 import org.opendaylight.yangtools.yang.common.RpcResult;
57 import org.slf4j.Logger;
58 import org.slf4j.LoggerFactory;
60 public class StatisticsProvider implements AutoCloseable {
62 public final static Logger spLogger = LoggerFactory.getLogger(StatisticsProvider.class);
64 private DataProviderService dps;
66 private NotificationProviderService nps;
68 private OpendaylightGroupStatisticsService groupStatsService;
70 private OpendaylightMeterStatisticsService meterStatsService;
72 private OpendaylightFlowStatisticsService flowStatsService;
74 private OpendaylightPortStatisticsService portStatsService;
76 private OpendaylightFlowTableStatisticsService flowTableStatsService;
78 private OpendaylightQueueStatisticsService queueStatsService;
80 private final MultipartMessageManager multipartMessageManager = new MultipartMessageManager();
82 private Thread statisticsRequesterThread;
84 private final InstanceIdentifier<Nodes> nodesIdentifier = InstanceIdentifier.builder(Nodes.class).toInstance();
86 private final int STATS_THREAD_EXECUTION_TIME= 50000;
87 //Local caching of stats
89 private final ConcurrentMap<NodeId,NodeStatistics> statisticsCache =
90 new ConcurrentHashMap<NodeId,NodeStatistics>();
92 public DataProviderService getDataService() {
96 public void setDataService(final DataProviderService dataService) {
97 this.dps = dataService;
100 public NotificationProviderService getNotificationService() {
104 public void setNotificationService(final NotificationProviderService notificationService) {
105 this.nps = notificationService;
108 public MultipartMessageManager getMultipartMessageManager() {
109 return multipartMessageManager;
112 private final StatisticsUpdateCommiter updateCommiter = new StatisticsUpdateCommiter(StatisticsProvider.this);
114 private Registration<NotificationListener> listenerRegistration;
116 public void start() {
118 NotificationProviderService nps = this.getNotificationService();
119 Registration<NotificationListener> registerNotificationListener = nps.registerNotificationListener(this.updateCommiter);
120 this.listenerRegistration = registerNotificationListener;
122 // Get Group/Meter statistics service instance
123 groupStatsService = StatisticsManagerActivator.getProviderContext().
124 getRpcService(OpendaylightGroupStatisticsService.class);
126 meterStatsService = StatisticsManagerActivator.getProviderContext().
127 getRpcService(OpendaylightMeterStatisticsService.class);
129 flowStatsService = StatisticsManagerActivator.getProviderContext().
130 getRpcService(OpendaylightFlowStatisticsService.class);
132 portStatsService = StatisticsManagerActivator.getProviderContext().
133 getRpcService(OpendaylightPortStatisticsService.class);
135 flowTableStatsService = StatisticsManagerActivator.getProviderContext().
136 getRpcService(OpendaylightFlowTableStatisticsService.class);
138 queueStatsService = StatisticsManagerActivator.getProviderContext().
139 getRpcService(OpendaylightQueueStatisticsService.class);
141 statisticsRequesterThread = new Thread( new Runnable(){
147 statsRequestSender();
149 Thread.sleep(STATS_THREAD_EXECUTION_TIME);
150 }catch (Exception e){
151 spLogger.error("Exception occurred while sending stats request : {}",e);
157 spLogger.debug("Statistics requester thread started with timer interval : {}",STATS_THREAD_EXECUTION_TIME);
159 statisticsRequesterThread.start();
161 spLogger.info("Statistics Provider started.");
164 protected DataModificationTransaction startChange() {
166 DataProviderService dps = this.getDataService();
167 return dps.beginTransaction();
170 private void statsRequestSender(){
172 List<Node> targetNodes = getAllConnectedNodes();
174 if(targetNodes == null)
178 for (Node targetNode : targetNodes){
180 InstanceIdentifier<Node> targetInstanceId = InstanceIdentifier.builder(Nodes.class).child(Node.class,targetNode.getKey()).toInstance();
181 NodeRef targetNodeRef = new NodeRef(targetInstanceId);
183 System.out.println("ANIL: Target Node object ::"+targetNode.toString());
185 System.out.println("ANIL: FlowCapableNode augmentations ::"+targetNode.getAugmentation(FlowCapableNode.class));
189 sendAggregateFlowsStatsFromAllTablesRequest(targetNode.getKey());
191 sendAllFlowsStatsFromAllTablesRequest(targetNodeRef);
193 sendAllPortStatisticsRequest(targetNodeRef);
195 sendAllFlowTablesStatisticsRequest(targetNodeRef);
197 sendAllQueueStatsFromAllNodeConnector (targetNodeRef);
200 spLogger.error("Exception occured while sending statistics requests : {}",e);
203 if(targetNode.getAugmentation(FlowCapableNode.class) != null){
205 spLogger.info("Send request for stats collection to node : {})",targetNode.getId());
208 sendAllGroupStatisticsRequest(targetNodeRef);
210 sendAllMeterStatisticsRequest(targetNodeRef);
212 sendGroupDescriptionRequest(targetNodeRef);
214 sendMeterConfigStatisticsRequest(targetNodeRef);
217 spLogger.error("Exception occured while sending statistics requests : {}", e);
223 private void sendAllFlowTablesStatisticsRequest(NodeRef targetNodeRef) throws InterruptedException, ExecutionException {
224 final GetFlowTablesStatisticsInputBuilder input =
225 new GetFlowTablesStatisticsInputBuilder();
227 input.setNode(targetNodeRef);
229 Future<RpcResult<GetFlowTablesStatisticsOutput>> response =
230 flowTableStatsService.getFlowTablesStatistics(input.build());
232 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
233 , StatsRequestType.ALL_FLOW_TABLE);
237 private void sendAllFlowsStatsFromAllTablesRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
238 final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input =
239 new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
241 input.setNode(targetNode);
243 Future<RpcResult<GetAllFlowsStatisticsFromAllFlowTablesOutput>> response =
244 flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build());
246 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
247 , StatsRequestType.ALL_FLOW);
251 private void sendAggregateFlowsStatsFromAllTablesRequest(NodeKey targetNodeKey) throws InterruptedException, ExecutionException{
253 List<Short> tablesId = getTablesFromNode(targetNodeKey);
255 if(tablesId.size() != 0){
256 for(Short id : tablesId){
258 spLogger.info("Send aggregate stats request for flow table {} to node {}",id,targetNodeKey);
259 GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
260 new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
262 input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
263 input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(id));
264 Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
265 flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());
267 multipartMessageManager.setTxIdAndTableIdMapEntry(response.get().getResult().getTransactionId(), id);
268 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
269 , StatsRequestType.AGGR_FLOW);
273 //Note: Just for testing, because i am not able to fetch table list from datastore
274 // Bug-225 is raised for investigation.
276 // spLogger.info("Send aggregate stats request for flow table {} to node {}",1,targetNodeKey);
277 // GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
278 // new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
280 // input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
281 // input.setTableId(new TableId((short)1));
282 // Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
283 // flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());`
285 // multipartMessageManager.setTxIdAndTableIdMapEntry(response.get().getResult().getTransactionId(), (short)1);
289 private void sendAllPortStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
291 final GetAllPortsStatisticsInputBuilder input = new GetAllPortsStatisticsInputBuilder();
293 input.setNode(targetNode);
295 Future<RpcResult<GetAllPortsStatisticsOutput>> response =
296 portStatsService.getAllPortsStatistics(input.build());
297 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
298 , StatsRequestType.ALL_PORT);
302 private void sendAllGroupStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
304 final GetAllGroupStatisticsInputBuilder input = new GetAllGroupStatisticsInputBuilder();
306 input.setNode(targetNode);
308 Future<RpcResult<GetAllGroupStatisticsOutput>> response =
309 groupStatsService.getAllGroupStatistics(input.build());
311 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
312 , StatsRequestType.ALL_GROUP);
316 private void sendGroupDescriptionRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
317 final GetGroupDescriptionInputBuilder input = new GetGroupDescriptionInputBuilder();
319 input.setNode(targetNode);
321 Future<RpcResult<GetGroupDescriptionOutput>> response =
322 groupStatsService.getGroupDescription(input.build());
324 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
325 , StatsRequestType.GROUP_DESC);
329 private void sendAllMeterStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
331 GetAllMeterStatisticsInputBuilder input = new GetAllMeterStatisticsInputBuilder();
333 input.setNode(targetNode);
335 Future<RpcResult<GetAllMeterStatisticsOutput>> response =
336 meterStatsService.getAllMeterStatistics(input.build());
338 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
339 , StatsRequestType.ALL_METER);;
343 private void sendMeterConfigStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
345 GetAllMeterConfigStatisticsInputBuilder input = new GetAllMeterConfigStatisticsInputBuilder();
347 input.setNode(targetNode);
349 Future<RpcResult<GetAllMeterConfigStatisticsOutput>> response =
350 meterStatsService.getAllMeterConfigStatistics(input.build());
352 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
353 , StatsRequestType.METER_CONFIG);;
357 private void sendAllQueueStatsFromAllNodeConnector(NodeRef targetNode) throws InterruptedException, ExecutionException {
358 GetAllQueuesStatisticsFromAllPortsInputBuilder input = new GetAllQueuesStatisticsFromAllPortsInputBuilder();
360 input.setNode(targetNode);
362 Future<RpcResult<GetAllQueuesStatisticsFromAllPortsOutput>> response =
363 queueStatsService.getAllQueuesStatisticsFromAllPorts(input.build());
365 this.multipartMessageManager.addTxIdToRequestTypeEntry(response.get().getResult().getTransactionId()
366 , StatsRequestType.ALL_QUEUE_STATS);;
370 public ConcurrentMap<NodeId, NodeStatistics> getStatisticsCache() {
371 return statisticsCache;
374 private List<Node> getAllConnectedNodes(){
376 Nodes nodes = (Nodes) dps.readOperationalData(nodesIdentifier);
380 spLogger.info("Number of connected nodes : {}",nodes.getNode().size());
381 return nodes.getNode();
384 private List<Short> getTablesFromNode(NodeKey nodeKey){
385 InstanceIdentifier<FlowCapableNode> nodesIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class,nodeKey).augmentation(FlowCapableNode.class).toInstance();
387 FlowCapableNode node = (FlowCapableNode)dps.readConfigurationData(nodesIdentifier);
388 List<Short> tablesId = new ArrayList<Short>();
389 if(node != null && node.getTable()!=null){
390 spLogger.info("Number of tables {} supported by node {}",node.getTable().size(),nodeKey);
391 for(Table table: node.getTable()){
392 tablesId.add(table.getId());
398 @SuppressWarnings("deprecation")
403 spLogger.info("Statistics Provider stopped.");
404 if (this.listenerRegistration != null) {
406 this.listenerRegistration.close();
408 this.statisticsRequesterThread.destroy();
411 } catch (Throwable e) {
412 throw Exceptions.sneakyThrow(e);