opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsProvider.java

   1 /*
   2  * Copyright IBM Corporation, 2013.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.md.statistics.manager;
   9
  10 import java.util.Collection;
  11 import java.util.Timer;
  12 import java.util.TimerTask;
  13 import java.util.concurrent.ConcurrentHashMap;
  14 import java.util.concurrent.ConcurrentMap;
  15 import java.util.concurrent.ExecutionException;
  16 import java.util.concurrent.Future;
  17 import java.util.concurrent.TimeUnit;
  18
  19 import org.opendaylight.controller.md.statistics.manager.MultipartMessageManager.StatsRequestType;
  20 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  21 import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
  22 import org.opendaylight.controller.sal.binding.api.data.DataBrokerService;
  23 import org.opendaylight.controller.sal.binding.api.data.DataChangeListener;
  24 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  25 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesOutput;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetFlowStatisticsFromFlowTableInputBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetFlowStatisticsFromFlowTableOutput;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsOutput;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.queue.rev130925.QueueId;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsInputBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsOutput;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionInputBuilder;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionOutput;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsService;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsInputBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsOutput;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsInputBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsOutput;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsService;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInputBuilder;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsOutput;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsInputBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsOutput;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetQueueStatisticsFromGivenPortInputBuilder;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetQueueStatisticsFromGivenPortOutput;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsService;
  70 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  71 import org.opendaylight.yangtools.concepts.Registration;
  72 import org.opendaylight.yangtools.yang.binding.DataObject;
  73 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  74 import org.opendaylight.yangtools.yang.binding.NotificationListener;
  75 import org.opendaylight.yangtools.yang.common.RpcResult;
  76 import org.slf4j.Logger;
  77 import org.slf4j.LoggerFactory;
  78
  79 import com.google.common.base.Preconditions;
  80
  81 /**
  82  * Following are main responsibilities of the class:
  83  * 1) Invoke statistics request thread to send periodic statistics request to all the
  84  * flow capable switch connected to the controller. It sends statistics request for
  85  * Group,Meter,Table,Flow,Queue,Aggregate stats.
  86  *
  87  * 2) Invoke statistics ager thread, to clean up all the stale statistics data from
  88  * operational data store.
  89  *
  90  * @author avishnoi@in.ibm.com
  91  *
  92  */
  93 public class StatisticsProvider implements AutoCloseable {
  94     public static final long STATS_COLLECTION_MILLIS = TimeUnit.SECONDS.toMillis(15);
  95
  96     private static final Logger spLogger = LoggerFactory.getLogger(StatisticsProvider.class);
  97
  98     private final ConcurrentMap<NodeId, NodeStatisticsHandler> handlers = new ConcurrentHashMap<>();
  99     private final MultipartMessageManager multipartMessageManager = new MultipartMessageManager();
 100     private final Timer timer = new Timer("statistics-manager", true);
 101     private final DataProviderService dps;
 102
 103     private OpendaylightGroupStatisticsService groupStatsService;
 104
 105     private OpendaylightMeterStatisticsService meterStatsService;
 106
 107     private OpendaylightFlowStatisticsService flowStatsService;
 108
 109     private OpendaylightPortStatisticsService portStatsService;
 110
 111     private OpendaylightFlowTableStatisticsService flowTableStatsService;
 112
 113     private OpendaylightQueueStatisticsService queueStatsService;
 114
 115     private StatisticsUpdateHandler statsUpdateHandler;
 116
 117     public StatisticsProvider(final DataProviderService dataService) {
 118         this.dps = Preconditions.checkNotNull(dataService);
 119     }
 120
 121     public MultipartMessageManager getMultipartMessageManager() {
 122         return multipartMessageManager;
 123     }
 124
 125     private final StatisticsListener updateCommiter = new StatisticsListener(StatisticsProvider.this);
 126
 127     private Registration<NotificationListener> listenerRegistration;
 128
 129     private ListenerRegistration<DataChangeListener> flowCapableTrackerRegistration;
 130
 131     public void start(final DataBrokerService dbs, final NotificationProviderService nps, final RpcConsumerRegistry rpcRegistry) {
 132
 133         // Get Group/Meter statistics service instances
 134         groupStatsService = rpcRegistry.getRpcService(OpendaylightGroupStatisticsService.class);
 135         meterStatsService = rpcRegistry.getRpcService(OpendaylightMeterStatisticsService.class);
 136         flowStatsService = rpcRegistry.getRpcService(OpendaylightFlowStatisticsService.class);
 137         portStatsService = rpcRegistry.getRpcService(OpendaylightPortStatisticsService.class);
 138         flowTableStatsService = rpcRegistry.getRpcService(OpendaylightFlowTableStatisticsService.class);
 139         queueStatsService = rpcRegistry.getRpcService(OpendaylightQueueStatisticsService.class);
 140
 141         // Start receiving notifications
 142         this.listenerRegistration = nps.registerNotificationListener(this.updateCommiter);
 143
 144         // Register for switch connect/disconnect notifications
 145         final InstanceIdentifier<FlowCapableNode> fcnId = InstanceIdentifier.builder(Nodes.class)
 146                 .child(Node.class).augmentation(FlowCapableNode.class).build();
 147         spLogger.debug("Registering FlowCapable tracker to {}", fcnId);
 148         this.flowCapableTrackerRegistration = dbs.registerDataChangeListener(fcnId,
 149                 new FlowCapableTracker(this, fcnId));
 150
 151         statsUpdateHandler = new StatisticsUpdateHandler(StatisticsProvider.this);
 152         registerDataStoreUpdateListener(dbs);
 153
 154         timer.schedule(new TimerTask() {
 155             @Override
 156             public void run() {
 157                 try {
 158                     // Send stats requests
 159                     for (NodeStatisticsHandler h : handlers.values()) {
 160                         sendStatisticsRequestsToNode(h);
 161                     }
 162
 163                     // Perform cleanup
 164                     for(NodeStatisticsHandler nodeStatisticsAger : handlers.values()){
 165                         nodeStatisticsAger.cleanStaleStatistics();
 166                     }
 167
 168                     multipartMessageManager.cleanStaleTransactionIds();
 169                 } catch (RuntimeException e) {
 170                     spLogger.warn("Failed to request statistics", e);
 171                 }
 172             }
 173         }, 0, STATS_COLLECTION_MILLIS);
 174
 175         spLogger.debug("Statistics timer task with timer interval : {}ms", STATS_COLLECTION_MILLIS);
 176         spLogger.info("Statistics Provider started.");
 177     }
 178
 179     private void registerDataStoreUpdateListener(DataBrokerService dbs) {
 180         // FIXME: the below should be broken out into StatisticsUpdateHandler
 181
 182         //Register for flow updates
 183         InstanceIdentifier<? extends DataObject> pathFlow = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 184                                                                     .augmentation(FlowCapableNode.class)
 185                                                                     .child(Table.class)
 186                                                                     .child(Flow.class).toInstance();
 187         dbs.registerDataChangeListener(pathFlow, statsUpdateHandler);
 188
 189         //Register for meter updates
 190         InstanceIdentifier<? extends DataObject> pathMeter = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 191                                                     .augmentation(FlowCapableNode.class)
 192                                                     .child(Meter.class).toInstance();
 193
 194         dbs.registerDataChangeListener(pathMeter, statsUpdateHandler);
 195
 196         //Register for group updates
 197         InstanceIdentifier<? extends DataObject> pathGroup = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 198                                                     .augmentation(FlowCapableNode.class)
 199                                                     .child(Group.class).toInstance();
 200         dbs.registerDataChangeListener(pathGroup, statsUpdateHandler);
 201
 202         //Register for queue updates
 203         InstanceIdentifier<? extends DataObject> pathQueue = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 204                                                                     .child(NodeConnector.class)
 205                                                                     .augmentation(FlowCapableNodeConnector.class)
 206                                                                     .child(Queue.class).toInstance();
 207         dbs.registerDataChangeListener(pathQueue, statsUpdateHandler);
 208     }
 209
 210     protected DataModificationTransaction startChange() {
 211         return dps.beginTransaction();
 212     }
 213
 214     private void sendStatisticsRequestsToNode(final NodeStatisticsHandler h) {
 215         NodeKey targetNode = h.getTargetNodeKey();
 216         spLogger.debug("Send requests for statistics collection to node : {}", targetNode.getId());
 217
 218         try{
 219             if(flowTableStatsService != null){
 220                 sendAllFlowTablesStatisticsRequest(h);
 221             }
 222             if(flowStatsService != null){
 223                 // FIXME: it does not make sense to trigger this before sendAllFlowTablesStatisticsRequest()
 224                 //        comes back -- we do not have any tables anyway.
 225                 sendAggregateFlowsStatsFromAllTablesRequest(h);
 226
 227                 sendAllFlowsStatsFromAllTablesRequest(h);
 228             }
 229             if(portStatsService != null){
 230                 sendAllNodeConnectorsStatisticsRequest(h);
 231             }
 232             if(groupStatsService != null){
 233                 sendAllGroupStatisticsRequest(h);
 234                 sendGroupDescriptionRequest(h);
 235             }
 236             if(meterStatsService != null){
 237                 sendAllMeterStatisticsRequest(h);
 238                 sendMeterConfigStatisticsRequest(h);
 239             }
 240             if(queueStatsService != null){
 241                 sendAllQueueStatsFromAllNodeConnector(h);
 242             }
 243         }catch(Exception e){
 244             spLogger.error("Exception occured while sending statistics requests : {}", e);
 245         }
 246     }
 247
 248
 249     private void sendAllFlowTablesStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException {
 250         final GetFlowTablesStatisticsInputBuilder input =
 251                 new GetFlowTablesStatisticsInputBuilder();
 252
 253         input.setNode(h.getTargetNodeRef());
 254
 255         Future<RpcResult<GetFlowTablesStatisticsOutput>> response =
 256                 flowTableStatsService.getFlowTablesStatistics(input.build());
 257
 258         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),response.get().getResult().getTransactionId()
 259                 , StatsRequestType.ALL_FLOW_TABLE);
 260
 261     }
 262
 263     private void sendAllFlowsStatsFromAllTablesRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 264         final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input =
 265                 new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
 266
 267         input.setNode(h.getTargetNodeRef());
 268
 269         Future<RpcResult<GetAllFlowsStatisticsFromAllFlowTablesOutput>> response =
 270                 flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build());
 271
 272         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 273                 , StatsRequestType.ALL_FLOW);
 274
 275     }
 276
 277     public void sendFlowStatsFromTableRequest(NodeKey node, Flow flow) throws InterruptedException, ExecutionException {
 278         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 279         if (h != null) {
 280             sendFlowStatsFromTableRequest(h, flow);
 281         }
 282     }
 283
 284     private void sendFlowStatsFromTableRequest(NodeStatisticsHandler h, Flow flow) throws InterruptedException, ExecutionException{
 285         final GetFlowStatisticsFromFlowTableInputBuilder input =
 286                 new GetFlowStatisticsFromFlowTableInputBuilder();
 287
 288         input.setNode(h.getTargetNodeRef());
 289         input.fieldsFrom(flow);
 290
 291         Future<RpcResult<GetFlowStatisticsFromFlowTableOutput>> response =
 292                 flowStatsService.getFlowStatisticsFromFlowTable(input.build());
 293
 294         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 295                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_FLOW);
 296     }
 297
 298     private void sendAggregateFlowsStatsFromAllTablesRequest(final NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 299         final Collection<TableKey> tables = h.getKnownTables();
 300         spLogger.debug("Node {} supports {} table(s)", h, tables.size());
 301
 302         for (TableKey key : h.getKnownTables()) {
 303             sendAggregateFlowsStatsFromTableRequest(h.getTargetNodeKey(), key.getId().shortValue());
 304         }
 305     }
 306
 307     private void sendAggregateFlowsStatsFromTableRequest(NodeKey targetNodeKey,Short tableId) throws InterruptedException, ExecutionException{
 308
 309         spLogger.debug("Send aggregate stats request for flow table {} to node {}",tableId,targetNodeKey);
 310         GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
 311                 new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
 312
 313         input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
 314         input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(tableId));
 315         Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
 316                 flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());
 317
 318         multipartMessageManager.setTxIdAndTableIdMapEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId(), tableId);
 319         this.multipartMessageManager.addTxIdToRequestTypeEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId()
 320                 , StatsRequestType.AGGR_FLOW);
 321     }
 322
 323     private void sendAllNodeConnectorsStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 324
 325         final GetAllNodeConnectorsStatisticsInputBuilder input = new GetAllNodeConnectorsStatisticsInputBuilder();
 326
 327         input.setNode(h.getTargetNodeRef());
 328
 329         Future<RpcResult<GetAllNodeConnectorsStatisticsOutput>> response =
 330                 portStatsService.getAllNodeConnectorsStatistics(input.build());
 331         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 332                 , StatsRequestType.ALL_PORT);
 333
 334     }
 335
 336     private void sendAllGroupStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 337
 338         final GetAllGroupStatisticsInputBuilder input = new GetAllGroupStatisticsInputBuilder();
 339
 340         input.setNode(h.getTargetNodeRef());
 341
 342         Future<RpcResult<GetAllGroupStatisticsOutput>> response =
 343                 groupStatsService.getAllGroupStatistics(input.build());
 344
 345         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 346                 , StatsRequestType.ALL_GROUP);
 347
 348     }
 349
 350     public void sendGroupDescriptionRequest(NodeKey node) throws InterruptedException, ExecutionException{
 351         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 352         if (h != null) {
 353             sendGroupDescriptionRequest(h);
 354         }
 355     }
 356
 357     private void sendGroupDescriptionRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 358         final GetGroupDescriptionInputBuilder input = new GetGroupDescriptionInputBuilder();
 359
 360         input.setNode(h.getTargetNodeRef());
 361
 362         Future<RpcResult<GetGroupDescriptionOutput>> response =
 363                 groupStatsService.getGroupDescription(input.build());
 364
 365         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 366                 response.get().getResult().getTransactionId(), StatsRequestType.GROUP_DESC);
 367     }
 368
 369     private void sendAllMeterStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 370
 371         GetAllMeterStatisticsInputBuilder input = new GetAllMeterStatisticsInputBuilder();
 372
 373         input.setNode(h.getTargetNodeRef());
 374
 375         Future<RpcResult<GetAllMeterStatisticsOutput>> response =
 376                 meterStatsService.getAllMeterStatistics(input.build());
 377
 378         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 379                 , StatsRequestType.ALL_METER);;
 380
 381     }
 382
 383     public void sendMeterConfigStatisticsRequest(NodeKey node) throws InterruptedException, ExecutionException {
 384         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 385         if (h != null) {
 386             sendMeterConfigStatisticsRequest(h);
 387         }
 388     }
 389
 390     private void sendMeterConfigStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 391
 392         GetAllMeterConfigStatisticsInputBuilder input = new GetAllMeterConfigStatisticsInputBuilder();
 393
 394         input.setNode(h.getTargetNodeRef());
 395
 396         Future<RpcResult<GetAllMeterConfigStatisticsOutput>> response =
 397                 meterStatsService.getAllMeterConfigStatistics(input.build());
 398
 399         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 400                 response.get().getResult().getTransactionId(), StatsRequestType.METER_CONFIG);;
 401     }
 402
 403     private void sendAllQueueStatsFromAllNodeConnector(NodeStatisticsHandler h) throws InterruptedException, ExecutionException {
 404         GetAllQueuesStatisticsFromAllPortsInputBuilder input = new GetAllQueuesStatisticsFromAllPortsInputBuilder();
 405
 406         input.setNode(h.getTargetNodeRef());
 407
 408         Future<RpcResult<GetAllQueuesStatisticsFromAllPortsOutput>> response =
 409                 queueStatsService.getAllQueuesStatisticsFromAllPorts(input.build());
 410
 411         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 412                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_QUEUE_STATS);;
 413     }
 414
 415     public void sendQueueStatsFromGivenNodeConnector(NodeKey node,NodeConnectorId nodeConnectorId, QueueId queueId) throws InterruptedException, ExecutionException {
 416         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 417         if (h != null) {
 418             sendQueueStatsFromGivenNodeConnector(h, nodeConnectorId, queueId);
 419         }
 420     }
 421
 422     private void sendQueueStatsFromGivenNodeConnector(NodeStatisticsHandler h, NodeConnectorId nodeConnectorId, QueueId queueId) throws InterruptedException, ExecutionException {
 423         GetQueueStatisticsFromGivenPortInputBuilder input = new GetQueueStatisticsFromGivenPortInputBuilder();
 424
 425         input.setNode(h.getTargetNodeRef());
 426         input.setNodeConnectorId(nodeConnectorId);
 427         input.setQueueId(queueId);
 428         Future<RpcResult<GetQueueStatisticsFromGivenPortOutput>> response =
 429                 queueStatsService.getQueueStatisticsFromGivenPort(input.build());
 430
 431         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 432                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_QUEUE_STATS);;
 433     }
 434
 435     /**
 436      * Get the handler for a particular node.
 437      *
 438      * @param nodeId source node
 439      * @return Node statistics handler for that node. Null if the statistics should
 440      *         not handled.
 441      */
 442     public final NodeStatisticsHandler getStatisticsHandler(final NodeId nodeId) {
 443         Preconditions.checkNotNull(nodeId);
 444         NodeStatisticsHandler handler = handlers.get(nodeId);
 445         if (handler == null) {
 446             spLogger.info("Attempted to get non-existing handler for {}", nodeId);
 447         }
 448         return handler;
 449     }
 450
 451     @Override
 452     public void close() {
 453         try {
 454             if (this.listenerRegistration != null) {
 455                 this.listenerRegistration.close();
 456                 this.listenerRegistration = null;
 457             }
 458             if (this.flowCapableTrackerRegistration != null) {
 459                 this.flowCapableTrackerRegistration.close();
 460                 this.flowCapableTrackerRegistration = null;
 461             }
 462             timer.cancel();
 463         } catch (Exception e) {
 464             spLogger.warn("Failed to stop Statistics Provider completely", e);
 465         } finally {
 466             spLogger.info("Statistics Provider stopped.");
 467         }
 468     }
 469
 470     void startNodeHandlers(final Collection<NodeKey> addedNodes) {
 471         for (NodeKey key : addedNodes) {
 472             if (handlers.containsKey(key.getId())) {
 473                 spLogger.warn("Attempted to start already-existing handler for {}, very strange", key.getId());
 474                 continue;
 475             }
 476
 477             final NodeStatisticsHandler h = new NodeStatisticsHandler(dps, key);
 478             final NodeStatisticsHandler old = handlers.putIfAbsent(key.getId(), h);
 479             if (old == null) {
 480                 spLogger.debug("Started node handler for {}", key.getId());
 481
 482                 // FIXME: this should be in the NodeStatisticsHandler itself
 483                 sendStatisticsRequestsToNode(h);
 484             } else {
 485                 spLogger.debug("Prevented race on handler for {}", key.getId());
 486             }
 487         }
 488     }
 489
 490     void stopNodeHandlers(final Collection<NodeKey> removedNodes) {
 491         for (NodeKey key : removedNodes) {
 492             final NodeStatisticsHandler s = handlers.remove(key.getId());
 493             if (s != null) {
 494                 spLogger.debug("Stopping node handler for {}", key.getId());
 495                 s.close();
 496             } else {
 497                 spLogger.warn("Attempted to remove non-existing handler for {}, very strange", key.getId());
 498             }
 499         }
 500     }
 501 }