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