+
+ private void statsRequestSender(){
+
+ List<Node> targetNodes = getAllConnectedNodes();
+
+ if(targetNodes == null)
+ return;
+
+
+ for (Node targetNode : targetNodes){
+
+ if(targetNode.getAugmentation(FlowCapableNode.class) != null){
+ sendStatisticsRequestsToNode(targetNode);
+ }
+ }
+ }
+
+ public void sendStatisticsRequestsToNode(Node targetNode){
+
+ spLogger.debug("Send requests for statistics collection to node : {})",targetNode.getId());
+
+ InstanceIdentifier<Node> targetInstanceId = InstanceIdentifier.builder(Nodes.class).child(Node.class,targetNode.getKey()).toInstance();
+
+ NodeRef targetNodeRef = new NodeRef(targetInstanceId);
+
+ try{
+ if(flowStatsService != null){
+ sendAggregateFlowsStatsFromAllTablesRequest(targetNode.getKey());
+ sendAllFlowsStatsFromAllTablesRequest(targetNodeRef);
+ }
+ if(flowTableStatsService != null){
+ sendAllFlowTablesStatisticsRequest(targetNodeRef);
+ }
+ if(portStatsService != null){
+ sendAllNodeConnectorsStatisticsRequest(targetNodeRef);
+ }
+ if(groupStatsService != null){
+ sendAllGroupStatisticsRequest(targetNodeRef);
+ sendGroupDescriptionRequest(targetNodeRef);
+ }
+ if(meterStatsService != null){
+ sendAllMeterStatisticsRequest(targetNodeRef);
+ sendMeterConfigStatisticsRequest(targetNodeRef);
+ }
+ if(queueStatsService != null){
+ sendAllQueueStatsFromAllNodeConnector (targetNodeRef);
+ }
+ }catch(Exception e){
+ spLogger.error("Exception occured while sending statistics requests : {}", e);
+ }
+ }
+
+
+ public void sendAllFlowTablesStatisticsRequest(NodeRef targetNodeRef) throws InterruptedException, ExecutionException {
+ final GetFlowTablesStatisticsInputBuilder input =
+ new GetFlowTablesStatisticsInputBuilder();
+
+ input.setNode(targetNodeRef);
+
+ Future<RpcResult<GetFlowTablesStatisticsOutput>> response =
+ flowTableStatsService.getFlowTablesStatistics(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNodeRef),response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_FLOW_TABLE);
+
+ }
+
+ public void sendAllFlowsStatsFromAllTablesRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+ final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder input =
+ new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllFlowsStatisticsFromAllFlowTablesOutput>> response =
+ flowStatsService.getAllFlowsStatisticsFromAllFlowTables(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_FLOW);
+
+ }
+
+ public void sendFlowStatsFromTableRequest(NodeRef targetNode,Flow flow) throws InterruptedException, ExecutionException{
+ final GetFlowStatisticsFromFlowTableInputBuilder input =
+ new GetFlowStatisticsFromFlowTableInputBuilder();
+
+ input.setNode(targetNode);
+ input.fieldsFrom(flow);
+
+ Future<RpcResult<GetFlowStatisticsFromFlowTableOutput>> response =
+ flowStatsService.getFlowStatisticsFromFlowTable(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_FLOW);
+
+ }
+
+ public void sendAggregateFlowsStatsFromAllTablesRequest(NodeKey targetNodeKey) throws InterruptedException, ExecutionException{
+
+ List<Short> tablesId = getTablesFromNode(targetNodeKey);
+
+ if(tablesId.size() != 0){
+ for(Short id : tablesId){
+
+ sendAggregateFlowsStatsFromTableRequest(targetNodeKey,id);
+ }
+ }else{
+ spLogger.debug("No details found in data store for flow tables associated with Node {}",targetNodeKey);
+ }
+ }
+
+ public void sendAggregateFlowsStatsFromTableRequest(NodeKey targetNodeKey,Short tableId) throws InterruptedException, ExecutionException{
+
+ spLogger.debug("Send aggregate stats request for flow table {} to node {}",tableId,targetNodeKey);
+ GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder input =
+ new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
+
+ input.setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).toInstance()));
+ input.setTableId(new org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId(tableId));
+ Future<RpcResult<GetAggregateFlowStatisticsFromFlowTableForAllFlowsOutput>> response =
+ flowStatsService.getAggregateFlowStatisticsFromFlowTableForAllFlows(input.build());
+
+ multipartMessageManager.setTxIdAndTableIdMapEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId(), tableId);
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(targetNodeKey.getId(), response.get().getResult().getTransactionId()
+ , StatsRequestType.AGGR_FLOW);
+ }
+
+ public void sendAllNodeConnectorsStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+
+ final GetAllNodeConnectorsStatisticsInputBuilder input = new GetAllNodeConnectorsStatisticsInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllNodeConnectorsStatisticsOutput>> response =
+ portStatsService.getAllNodeConnectorsStatistics(input.build());
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_PORT);
+
+ }
+
+ public void sendAllGroupStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+
+ final GetAllGroupStatisticsInputBuilder input = new GetAllGroupStatisticsInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllGroupStatisticsOutput>> response =
+ groupStatsService.getAllGroupStatistics(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_GROUP);
+
+ }
+
+ public void sendGroupDescriptionRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+ final GetGroupDescriptionInputBuilder input = new GetGroupDescriptionInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetGroupDescriptionOutput>> response =
+ groupStatsService.getGroupDescription(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.GROUP_DESC);
+
+ }
+
+ public void sendAllMeterStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+
+ GetAllMeterStatisticsInputBuilder input = new GetAllMeterStatisticsInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllMeterStatisticsOutput>> response =
+ meterStatsService.getAllMeterStatistics(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_METER);;
+
+ }
+
+ public void sendMeterConfigStatisticsRequest(NodeRef targetNode) throws InterruptedException, ExecutionException{
+
+ GetAllMeterConfigStatisticsInputBuilder input = new GetAllMeterConfigStatisticsInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllMeterConfigStatisticsOutput>> response =
+ meterStatsService.getAllMeterConfigStatistics(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.METER_CONFIG);;
+
+ }
+
+ public void sendAllQueueStatsFromAllNodeConnector(NodeRef targetNode) throws InterruptedException, ExecutionException {
+ GetAllQueuesStatisticsFromAllPortsInputBuilder input = new GetAllQueuesStatisticsFromAllPortsInputBuilder();
+
+ input.setNode(targetNode);
+
+ Future<RpcResult<GetAllQueuesStatisticsFromAllPortsOutput>> response =
+ queueStatsService.getAllQueuesStatisticsFromAllPorts(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_QUEUE_STATS);;
+
+ }
+
+ public void sendQueueStatsFromGivenNodeConnector(NodeRef targetNode,NodeConnectorId nodeConnectorId, QueueId queueId) throws InterruptedException, ExecutionException {
+ GetQueueStatisticsFromGivenPortInputBuilder input = new GetQueueStatisticsFromGivenPortInputBuilder();
+
+ input.setNode(targetNode);
+ input.setNodeConnectorId(nodeConnectorId);
+ input.setQueueId(queueId);
+ Future<RpcResult<GetQueueStatisticsFromGivenPortOutput>> response =
+ queueStatsService.getQueueStatisticsFromGivenPort(input.build());
+
+ this.multipartMessageManager.addTxIdToRequestTypeEntry(getNodeId(targetNode), response.get().getResult().getTransactionId()
+ , StatsRequestType.ALL_QUEUE_STATS);;
+
+ }
+
+ /**
+ * Get the handler for a particular node.
+ *
+ * @param nodeId source node
+ * @return Node statistics handler for that node. Null if the statistics should
+ * not handled.
+ */
+ public final NodeStatisticsAger getStatisticsHandler(final NodeId nodeId) {
+ Preconditions.checkNotNull(nodeId);
+ NodeStatisticsAger ager = statisticsCache.get(nodeId);
+ if (ager == null) {
+ ager = new NodeStatisticsAger(this, new NodeKey(nodeId));
+ statisticsCache.put(nodeId, ager);
+ }
+
+ return ager;
+ }
+
+ private List<Node> getAllConnectedNodes(){
+ Nodes nodes = (Nodes) dps.readOperationalData(nodesIdentifier);
+ if(nodes == null)
+ return null;
+
+ spLogger.debug("Number of connected nodes : {}",nodes.getNode().size());
+ return nodes.getNode();
+ }
+
+ private List<Short> getTablesFromNode(NodeKey nodeKey){
+ InstanceIdentifier<FlowCapableNode> nodesIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class,nodeKey).augmentation(FlowCapableNode.class).toInstance();
+
+ FlowCapableNode node = (FlowCapableNode)dps.readOperationalData(nodesIdentifier);
+ List<Short> tablesId = new ArrayList<Short>();
+ if(node != null && node.getTable()!=null){
+ spLogger.debug("Number of tables {} supported by node {}",node.getTable().size(),nodeKey);
+ for(Table table: node.getTable()){
+ tablesId.add(table.getId());
+ }
+ }
+ return tablesId;
+ }
+
+ @SuppressWarnings("unchecked")
+ private NodeId getNodeId(NodeRef nodeRef){
+ InstanceIdentifier<Node> nodeII = (InstanceIdentifier<Node>) nodeRef.getValue();
+ NodeKey nodeKey = InstanceIdentifier.keyOf(nodeII);
+ return nodeKey.getId();
+ }
+
+ @SuppressWarnings("deprecation")