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                     statsRequestSender();
 160
 161                     // Perform cleanup
 162                     for(NodeStatisticsHandler nodeStatisticsAger : handlers.values()){
 163                         nodeStatisticsAger.cleanStaleStatistics();
 164                     }
 165
 166                     multipartMessageManager.cleanStaleTransactionIds();
 167                 } catch (RuntimeException e) {
 168                     spLogger.warn("Failed to request statistics", e);
 169                 }
 170             }
 171         }, 0, STATS_COLLECTION_MILLIS);
 172
 173         spLogger.debug("Statistics timer task with timer interval : {}ms", STATS_COLLECTION_MILLIS);
 174         spLogger.info("Statistics Provider started.");
 175     }
 176
 177     private void registerDataStoreUpdateListener(DataBrokerService dbs) {
 178         // FIXME: the below should be broken out into StatisticsUpdateHandler
 179
 180         //Register for flow updates
 181         InstanceIdentifier<? extends DataObject> pathFlow = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 182                                                                     .augmentation(FlowCapableNode.class)
 183                                                                     .child(Table.class)
 184                                                                     .child(Flow.class).toInstance();
 185         dbs.registerDataChangeListener(pathFlow, statsUpdateHandler);
 186
 187         //Register for meter updates
 188         InstanceIdentifier<? extends DataObject> pathMeter = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 189                                                     .augmentation(FlowCapableNode.class)
 190                                                     .child(Meter.class).toInstance();
 191
 192         dbs.registerDataChangeListener(pathMeter, statsUpdateHandler);
 193
 194         //Register for group updates
 195         InstanceIdentifier<? extends DataObject> pathGroup = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 196                                                     .augmentation(FlowCapableNode.class)
 197                                                     .child(Group.class).toInstance();
 198         dbs.registerDataChangeListener(pathGroup, statsUpdateHandler);
 199
 200         //Register for queue updates
 201         InstanceIdentifier<? extends DataObject> pathQueue = InstanceIdentifier.builder(Nodes.class).child(Node.class)
 202                                                                     .child(NodeConnector.class)
 203                                                                     .augmentation(FlowCapableNodeConnector.class)
 204                                                                     .child(Queue.class).toInstance();
 205         dbs.registerDataChangeListener(pathQueue, statsUpdateHandler);
 206     }
 207
 208     protected DataModificationTransaction startChange() {
 209         return dps.beginTransaction();
 210     }
 211
 212     private void statsRequestSender() {
 213         for (NodeStatisticsHandler h : handlers.values()) {
 214             sendStatisticsRequestsToNode(h);
 215         }
 216     }
 217
 218     private void sendStatisticsRequestsToNode(final NodeStatisticsHandler h) {
 219         NodeKey targetNode = h.getTargetNodeKey();
 220         spLogger.debug("Send requests for statistics collection to node : {}", targetNode.getId());
 221
 222         try{
 223             if(flowTableStatsService != null){
 224                 sendAllFlowTablesStatisticsRequest(h);
 225             }
 226             if(flowStatsService != null){
 227                 // FIXME: it does not make sense to trigger this before sendAllFlowTablesStatisticsRequest()
 228                 //        comes back -- we do not have any tables anyway.
 229                 sendAggregateFlowsStatsFromAllTablesRequest(h);
 230
 231                 sendAllFlowsStatsFromAllTablesRequest(h);
 232             }
 233             if(portStatsService != null){
 234                 sendAllNodeConnectorsStatisticsRequest(h);
 235             }
 236             if(groupStatsService != null){
 237                 sendAllGroupStatisticsRequest(h);
 238                 sendGroupDescriptionRequest(h);
 239             }
 240             if(meterStatsService != null){
 241                 sendAllMeterStatisticsRequest(h);
 242                 sendMeterConfigStatisticsRequest(h);
 243             }
 244             if(queueStatsService != null){
 245                 sendAllQueueStatsFromAllNodeConnector(h);
 246             }
 247         }catch(Exception e){
 248             spLogger.error("Exception occured while sending statistics requests : {}", e);
 249         }
 250     }
 251
 252
 253     private void sendAllFlowTablesStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException {
 254         final GetFlowTablesStatisticsInputBuilder input =
 255                 new GetFlowTablesStatisticsInputBuilder();
 256
 257         input.setNode(h.getTargetNodeRef());
 258
 259         Future<RpcResult<GetFlowTablesStatisticsOutput>> response =
 260                 flowTableStatsService.getFlowTablesStatistics(input.build());
 261
 262         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),response.get().getResult().getTransactionId()
 263                 , StatsRequestType.ALL_FLOW_TABLE);
 264
 265     }
 266
 267     private void sendAllFlowsStatsFromAllTablesRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 268         final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input =
 269                 new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
 270
 271         input.setNode(h.getTargetNodeRef());
 272
 273         Future<RpcResult<GetAllFlowsStatisticsFromAllFlowTablesOutput>> response =
 274                 flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build());
 275
 276         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 277                 , StatsRequestType.ALL_FLOW);
 278
 279     }
 280
 281     public void sendFlowStatsFromTableRequest(NodeKey node, Flow flow) throws InterruptedException, ExecutionException {
 282         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 283         if (h != null) {
 284             sendFlowStatsFromTableRequest(h, flow);
 285         }
 286     }
 287
 288     private void sendFlowStatsFromTableRequest(NodeStatisticsHandler h, Flow flow) throws InterruptedException, ExecutionException{
 289         final GetFlowStatisticsFromFlowTableInputBuilder input =
 290                 new GetFlowStatisticsFromFlowTableInputBuilder();
 291
 292         input.setNode(h.getTargetNodeRef());
 293         input.fieldsFrom(flow);
 294
 295         Future<RpcResult<GetFlowStatisticsFromFlowTableOutput>> response =
 296                 flowStatsService.getFlowStatisticsFromFlowTable(input.build());
 297
 298         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 299                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_FLOW);
 300     }
 301
 302     private void sendAggregateFlowsStatsFromAllTablesRequest(final NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 303         final Collection<TableKey> tables = h.getKnownTables();
 304         spLogger.debug("Node {} supports {} table(s)", h, tables.size());
 305
 306         for (TableKey key : h.getKnownTables()) {
 307             sendAggregateFlowsStatsFromTableRequest(h.getTargetNodeKey(), key.getId().shortValue());
 308         }
 309     }
 310
 311     private void sendAggregateFlowsStatsFromTableRequest(NodeKey targetNodeKey,Short tableId) throws InterruptedException, ExecutionException{
 312
 313         spLogger.debug("Send aggregate stats request for flow table {} to node {}",tableId,targetNodeKey);
 314         GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
 315                 new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
 316
 317         input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
 318         input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(tableId));
 319         Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
 320                 flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());
 321
 322         multipartMessageManager.setTxIdAndTableIdMapEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId(), tableId);
 323         this.multipartMessageManager.addTxIdToRequestTypeEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId()
 324                 , StatsRequestType.AGGR_FLOW);
 325     }
 326
 327     private void sendAllNodeConnectorsStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 328
 329         final GetAllNodeConnectorsStatisticsInputBuilder input = new GetAllNodeConnectorsStatisticsInputBuilder();
 330
 331         input.setNode(h.getTargetNodeRef());
 332
 333         Future<RpcResult<GetAllNodeConnectorsStatisticsOutput>> response =
 334                 portStatsService.getAllNodeConnectorsStatistics(input.build());
 335         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 336                 , StatsRequestType.ALL_PORT);
 337
 338     }
 339
 340     private void sendAllGroupStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 341
 342         final GetAllGroupStatisticsInputBuilder input = new GetAllGroupStatisticsInputBuilder();
 343
 344         input.setNode(h.getTargetNodeRef());
 345
 346         Future<RpcResult<GetAllGroupStatisticsOutput>> response =
 347                 groupStatsService.getAllGroupStatistics(input.build());
 348
 349         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 350                 , StatsRequestType.ALL_GROUP);
 351
 352     }
 353
 354     public void sendGroupDescriptionRequest(NodeKey node) throws InterruptedException, ExecutionException{
 355         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 356         if (h != null) {
 357             sendGroupDescriptionRequest(h);
 358         }
 359     }
 360
 361     private void sendGroupDescriptionRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 362         final GetGroupDescriptionInputBuilder input = new GetGroupDescriptionInputBuilder();
 363
 364         input.setNode(h.getTargetNodeRef());
 365
 366         Future<RpcResult<GetGroupDescriptionOutput>> response =
 367                 groupStatsService.getGroupDescription(input.build());
 368
 369         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 370                 response.get().getResult().getTransactionId(), StatsRequestType.GROUP_DESC);
 371     }
 372
 373     private void sendAllMeterStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 374
 375         GetAllMeterStatisticsInputBuilder input = new GetAllMeterStatisticsInputBuilder();
 376
 377         input.setNode(h.getTargetNodeRef());
 378
 379         Future<RpcResult<GetAllMeterStatisticsOutput>> response =
 380                 meterStatsService.getAllMeterStatistics(input.build());
 381
 382         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(), response.get().getResult().getTransactionId()
 383                 , StatsRequestType.ALL_METER);;
 384
 385     }
 386
 387     public void sendMeterConfigStatisticsRequest(NodeKey node) throws InterruptedException, ExecutionException {
 388         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 389         if (h != null) {
 390             sendMeterConfigStatisticsRequest(h);
 391         }
 392     }
 393
 394     private void sendMeterConfigStatisticsRequest(NodeStatisticsHandler h) throws InterruptedException, ExecutionException{
 395
 396         GetAllMeterConfigStatisticsInputBuilder input = new GetAllMeterConfigStatisticsInputBuilder();
 397
 398         input.setNode(h.getTargetNodeRef());
 399
 400         Future<RpcResult<GetAllMeterConfigStatisticsOutput>> response =
 401                 meterStatsService.getAllMeterConfigStatistics(input.build());
 402
 403         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 404                 response.get().getResult().getTransactionId(), StatsRequestType.METER_CONFIG);;
 405     }
 406
 407     private void sendAllQueueStatsFromAllNodeConnector(NodeStatisticsHandler h) throws InterruptedException, ExecutionException {
 408         GetAllQueuesStatisticsFromAllPortsInputBuilder input = new GetAllQueuesStatisticsFromAllPortsInputBuilder();
 409
 410         input.setNode(h.getTargetNodeRef());
 411
 412         Future<RpcResult<GetAllQueuesStatisticsFromAllPortsOutput>> response =
 413                 queueStatsService.getAllQueuesStatisticsFromAllPorts(input.build());
 414
 415         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 416                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_QUEUE_STATS);;
 417     }
 418
 419     public void sendQueueStatsFromGivenNodeConnector(NodeKey node,NodeConnectorId nodeConnectorId, QueueId queueId) throws InterruptedException, ExecutionException {
 420         final NodeStatisticsHandler h = getStatisticsHandler(node.getId());
 421         if (h != null) {
 422             sendQueueStatsFromGivenNodeConnector(h, nodeConnectorId, queueId);
 423         }
 424     }
 425
 426     private void sendQueueStatsFromGivenNodeConnector(NodeStatisticsHandler h, NodeConnectorId nodeConnectorId, QueueId queueId) throws InterruptedException, ExecutionException {
 427         GetQueueStatisticsFromGivenPortInputBuilder input = new GetQueueStatisticsFromGivenPortInputBuilder();
 428
 429         input.setNode(h.getTargetNodeRef());
 430         input.setNodeConnectorId(nodeConnectorId);
 431         input.setQueueId(queueId);
 432         Future<RpcResult<GetQueueStatisticsFromGivenPortOutput>> response =
 433                 queueStatsService.getQueueStatisticsFromGivenPort(input.build());
 434
 435         this.multipartMessageManager.addTxIdToRequestTypeEntry(h.getTargetNodeKey().getId(),
 436                 response.get().getResult().getTransactionId(), StatsRequestType.ALL_QUEUE_STATS);;
 437     }
 438
 439     /**
 440      * Get the handler for a particular node.
 441      *
 442      * @param nodeId source node
 443      * @return Node statistics handler for that node. Null if the statistics should
 444      *         not handled.
 445      */
 446     public final NodeStatisticsHandler getStatisticsHandler(final NodeId nodeId) {
 447         Preconditions.checkNotNull(nodeId);
 448         NodeStatisticsHandler handler = handlers.get(nodeId);
 449         if (handler == null) {
 450             spLogger.info("Attempted to get non-existing handler for {}", nodeId);
 451         }
 452         return handler;
 453     }
 454
 455     @Override
 456     public void close() {
 457         try {
 458             if (this.listenerRegistration != null) {
 459                 this.listenerRegistration.close();
 460                 this.listenerRegistration = null;
 461             }
 462             if (this.flowCapableTrackerRegistration != null) {
 463                 this.flowCapableTrackerRegistration.close();
 464                 this.flowCapableTrackerRegistration = null;
 465             }
 466             timer.cancel();
 467         } catch (Exception e) {
 468             spLogger.warn("Failed to stop Statistics Provider completely", e);
 469         } finally {
 470             spLogger.info("Statistics Provider stopped.");
 471         }
 472     }
 473
 474     void startNodeHandlers(final Collection<NodeKey> addedNodes) {
 475         for (NodeKey key : addedNodes) {
 476             if (handlers.containsKey(key.getId())) {
 477                 spLogger.warn("Attempted to start already-existing handler for {}, very strange", key.getId());
 478                 continue;
 479             }
 480
 481             final NodeStatisticsHandler h = new NodeStatisticsHandler(dps, key);
 482             final NodeStatisticsHandler old = handlers.putIfAbsent(key.getId(), h);
 483             if (old == null) {
 484                 spLogger.debug("Started node handler for {}", key.getId());
 485
 486                 // FIXME: this should be in the NodeStatisticsHandler itself
 487                 sendStatisticsRequestsToNode(h);
 488             } else {
 489                 spLogger.debug("Prevented race on handler for {}", key.getId());
 490             }
 491         }
 492     }
 493
 494     void stopNodeHandlers(final Collection<NodeKey> removedNodes) {
 495         for (NodeKey key : removedNodes) {
 496             final NodeStatisticsHandler s = handlers.remove(key.getId());
 497             if (s != null) {
 498                 spLogger.debug("Stopping node handler for {}", key.getId());
 499                 s.close();
 500             } else {
 501                 spLogger.warn("Attempted to remove non-existing handler for {}, very strange", key.getId());
 502             }
 503         }
 504     }
 505 }