interfacemgr/interfacemgr-impl/src/main/java/org/opendaylight/vpnservice/interfacemgr/pmcounters/NodeConnectorStatsImpl.java

   1 /*
   2  * Copyright (c) 2016 Ericsson India Global Services Pvt Ltd. and others.  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.vpnservice.interfacemgr.pmcounters;
   9
  10 import java.lang.Thread.UncaughtExceptionHandler;
  11 import java.math.BigInteger;
  12 import java.util.ArrayList;
  13 import java.util.HashMap;
  14 import java.util.List;
  15 import java.util.Map;
  16 import java.util.concurrent.ConcurrentHashMap;
  17 import java.util.concurrent.Executors;
  18 import java.util.concurrent.ScheduledExecutorService;
  19 import java.util.concurrent.ScheduledFuture;
  20 import java.util.concurrent.ThreadFactory;
  21 import java.util.concurrent.TimeUnit;
  22
  23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  24 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  25 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.vpnservice.mdsalutil.AbstractDataChangeListener;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.FlowTableStatisticsUpdate;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInput;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsListener;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInput;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInputBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.NodeConnectorStatisticsUpdate;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsListener;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
  46 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  51
  52 public class NodeConnectorStatsImpl extends AbstractDataChangeListener<Node>{
  53
  54     private static final Logger logger = LoggerFactory.getLogger(NodeConnectorStatsImpl.class);
  55     private static final int THREAD_POOL_SIZE = 4;
  56     private static final int NO_DELAY = 0;
  57     public static final PMAgentForNodeConnectorCounters pmagent = new PMAgentForNodeConnectorCounters();
  58     private PortRpcStatisticsListener portStatsListener = new PortRpcStatisticsListener();
  59     private FlowRpcStatisticsListener flowTableStatsListener = new FlowRpcStatisticsListener();
  60     private List<BigInteger> nodes = new ArrayList<>();
  61     Map<String, Map<String, String>> nodeAndNcIdOFPortDurationMap = new ConcurrentHashMap<String, Map<String, String>>();
  62     Map<String, Map<String, String>> nodeAndNcIdOFPortReceiveDropMap = new ConcurrentHashMap<String, Map<String, String>>();
  63     Map<String, Map<String, String>> nodeAndNcIdOFPortReceiveError = new ConcurrentHashMap<String, Map<String, String>>();
  64     Map<String, Map<String, String>> nodeAndNcIdPacketSentMap = new ConcurrentHashMap<String, Map<String, String>>();
  65     Map<String, Map<String, String>> nodeAndNcIdPacketReceiveMap = new ConcurrentHashMap<String, Map<String, String>>();
  66     Map<String, Map<String, String>> nodeAndNcIdBytesSentMap = new ConcurrentHashMap<String, Map<String, String>>();
  67     Map<String, Map<String, String>> nodeAndNcIdBytesReceiveMap = new ConcurrentHashMap<String, Map<String, String>>();
  68     Map<String, Map<String, String>> nodeAndEntriesPerOFTableMap = new ConcurrentHashMap<String, Map<String, String>>();
  69     private ScheduledFuture<?> scheduledResult;
  70     private OpendaylightPortStatisticsService statPortService;
  71     private ScheduledExecutorService portStatExecutorService;
  72     private OpendaylightFlowTableStatisticsService opendaylightFlowTableStatisticsService;
  73     public NodeConnectorStatsImpl(DataBroker db, NotificationService notificationService, OpendaylightPortStatisticsService statPortService, OpendaylightFlowTableStatisticsService opendaylightFlowTableStatisticsService) {
  74         super(Node.class);
  75         this.statPortService = statPortService;
  76         this.opendaylightFlowTableStatisticsService = opendaylightFlowTableStatisticsService;
  77         registerListener(db);
  78         portStatExecutorService = Executors.newScheduledThreadPool(THREAD_POOL_SIZE, getThreadFactory("Port Stats Request Task"));
  79         notificationService.registerNotificationListener(portStatsListener);
  80         notificationService.registerNotificationListener(flowTableStatsListener);
  81         pmagent.registerMbean();
  82     }
  83
  84     private void registerListener(final DataBroker db) {
  85         try {
  86             db.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL,
  87                     getWildCardPath(), NodeConnectorStatsImpl.this, AsyncDataBroker.DataChangeScope.SUBTREE);
  88         } catch (final Exception e) {
  89             logger.error("NodeConnectorStatsImpl: DataChange listener registration fail!", e);
  90             throw new IllegalStateException("NodeConnectorStatsImpl: registration Listener failed.", e);
  91         }
  92     }
  93
  94     private InstanceIdentifier<Node> getWildCardPath() {
  95         return InstanceIdentifier.create(Nodes.class).child(Node.class);
  96     }
  97
  98     /*
  99      * PortStat request task is started when first DPN gets connected
 100      */
 101     private void schedulePortStatRequestTask() {
 102         logger.info("Scheduling port statistics request");
 103         PortStatRequestTask portStatRequestTask = new PortStatRequestTask();
 104         scheduledResult = portStatExecutorService.scheduleAtFixedRate(portStatRequestTask, NO_DELAY, 10000, TimeUnit.MILLISECONDS);
 105     }
 106
 107     /*
 108      * PortStat request task is stopped when last DPN is removed.
 109      */
 110     private void stopPortStatRequestTask() {
 111         if(scheduledResult != null) {
 112             logger.info("Stopping port statistics request");
 113             scheduledResult.cancel(true);
 114         }
 115     }
 116
 117     /*
 118      * This task queries for node connector statistics as well as flowtables statistics every 10 secs.
 119      * Minimum period which can be configured for PMJob is 10 secs.
 120      */
 121     private class PortStatRequestTask implements Runnable {
 122
 123         @Override
 124         public void run() {
 125             if(logger.isTraceEnabled()) {
 126                 logger.trace("Requesting port stats - {}");
 127             }
 128             for (BigInteger node : nodes) {
 129                 logger.trace("Requesting AllNodeConnectorStatistics for node - {}", node);
 130                 statPortService.getAllNodeConnectorsStatistics(buildGetAllNodeConnectorStatistics(node));
 131                 opendaylightFlowTableStatisticsService.getFlowTablesStatistics(buildGetFlowTablesStatistics(node));
 132             }
 133         }
 134
 135         private GetAllNodeConnectorsStatisticsInput buildGetAllNodeConnectorStatistics(BigInteger dpId) {
 136             return new GetAllNodeConnectorsStatisticsInputBuilder()
 137                     .setNode(
 138                             new NodeRef(InstanceIdentifier.builder(Nodes.class)
 139                                     .child(Node.class, new NodeKey(new NodeId("openflow:" + dpId.toString()))).build())).build();
 140         }
 141
 142         private GetFlowTablesStatisticsInput buildGetFlowTablesStatistics(BigInteger dpId) {
 143             return new GetFlowTablesStatisticsInputBuilder()
 144                     .setNode(
 145                             new NodeRef(InstanceIdentifier.builder(Nodes.class)
 146                                     .child(Node.class, new NodeKey(new NodeId("openflow:" + dpId.toString()))).build())).build();
 147         }
 148
 149     }
 150
 151     private ThreadFactory getThreadFactory(String threadNameFormat) {
 152         ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
 153         builder.setNameFormat(threadNameFormat);
 154         builder.setUncaughtExceptionHandler( new UncaughtExceptionHandler() {
 155             @Override
 156             public void uncaughtException(Thread t, Throwable e) {
 157                 logger.error("Received Uncaught Exception event in Thread: {}", t.getName(), e);
 158             }
 159         });
 160         return builder.build();
 161     }
 162
 163     /*
 164      * PortRpcStatisticsListener listens for the NodeConnectorStatisticsUpdate and then update the corresponding counter map
 165      */
 166     class PortRpcStatisticsListener implements OpendaylightPortStatisticsListener {
 167
 168         @Override
 169         public void onNodeConnectorStatisticsUpdate(NodeConnectorStatisticsUpdate ncStats) {
 170             Map<String, String> ncIdOFPortDurationMap = new HashMap<String, String>();
 171             Map<String, String> ncIdOFPortReceiveDropMap = new HashMap<String, String>();
 172             Map<String, String> ncIdOFPortReceiveError = new HashMap<String, String>();
 173             Map<String, String> ncIdPacketSentMap = new HashMap<String, String>();
 174             Map<String, String> ncIdPacketReceiveMap = new HashMap<String, String>();
 175             Map<String, String> ncIdBytesSentMap = new HashMap<String, String>();
 176             Map<String, String> ncIdBytesReceiveMap = new HashMap<String, String>();
 177             List<NodeConnectorStatisticsAndPortNumberMap> ncStatsAndPortMapList = ncStats.getNodeConnectorStatisticsAndPortNumberMap();
 178             NodeId nodeId = ncStats.getId();
 179             String node = nodeId.getValue().split(":")[1];
 180             for (NodeConnectorStatisticsAndPortNumberMap ncStatsAndPortMap : ncStatsAndPortMapList) {
 181                 NodeConnectorId nodeConnector = ncStatsAndPortMap.getNodeConnectorId();
 182                 String port = nodeConnector.getValue().split(":")[2];
 183                 String nodePortStr = "dpnId_" + node + "_portNum_" + port;
 184                 ncIdOFPortDurationMap.put("OFPortDuration:" + nodePortStr + "_OFPortDuration", ncStatsAndPortMap.getDuration().getSecond().getValue().toString());
 185                 ncIdOFPortReceiveDropMap.put("PacketsPerOFPortReceiveDrop:" + nodePortStr + "_PacketsPerOFPortReceiveDrop", ncStatsAndPortMap.getReceiveDrops().toString());
 186                 ncIdOFPortReceiveError.put("PacketsPerOFPortReceiveError:" + nodePortStr + "_PacketsPerOFPortReceiveError", ncStatsAndPortMap.getReceiveErrors().toString());
 187                 ncIdPacketSentMap.put("PacketsPerOFPortSent:" + nodePortStr + "_PacketsPerOFPortSent", ncStatsAndPortMap.getPackets().getTransmitted().toString());
 188                 ncIdPacketReceiveMap.put("PacketsPerOFPortReceive:" + nodePortStr + "_PacketsPerOFPortReceive", ncStatsAndPortMap.getPackets().getReceived().toString());
 189                 ncIdBytesSentMap.put("BytesPerOFPortSent:" + nodePortStr + "_BytesPerOFPortSent", ncStatsAndPortMap.getBytes().getTransmitted().toString());
 190                 ncIdBytesReceiveMap.put("BytesPerOFPortReceive:" + nodePortStr + "_BytesPerOFPortReceive", ncStatsAndPortMap.getBytes().getReceived().toString());
 191             }
 192             logger.trace("Port Stats {}", ncStatsAndPortMapList);
 193             //Storing allNodeConnectorStats(like ncIdOFPortDurationMap) in a map with key as node for easy removal and addition of allNodeConnectorStats.
 194             nodeAndNcIdOFPortDurationMap.put(node, ncIdOFPortDurationMap);
 195             nodeAndNcIdOFPortReceiveDropMap.put(node, ncIdOFPortReceiveDropMap);
 196             nodeAndNcIdOFPortReceiveError.put(node, ncIdOFPortReceiveError);
 197             nodeAndNcIdPacketSentMap.put(node, ncIdPacketSentMap);
 198             nodeAndNcIdPacketReceiveMap.put(node, ncIdPacketReceiveMap);
 199             nodeAndNcIdBytesSentMap.put(node, ncIdBytesSentMap);
 200             nodeAndNcIdBytesReceiveMap.put(node, ncIdBytesReceiveMap);
 201             //Combining the stats of all nodeconnectors in all nodes. This Map will be stored under MBean which will be queried as regular intervals.
 202             ncIdOFPortDurationMap = combineAllNodesStats(nodeAndNcIdOFPortDurationMap);
 203             ncIdOFPortReceiveDropMap = combineAllNodesStats(nodeAndNcIdOFPortReceiveDropMap);
 204             ncIdOFPortReceiveError = combineAllNodesStats(nodeAndNcIdOFPortReceiveError);
 205             ncIdPacketSentMap = combineAllNodesStats(nodeAndNcIdPacketSentMap);
 206             ncIdPacketReceiveMap = combineAllNodesStats(nodeAndNcIdPacketReceiveMap);
 207             ncIdBytesSentMap = combineAllNodesStats(nodeAndNcIdBytesSentMap);
 208             ncIdBytesReceiveMap = combineAllNodesStats(nodeAndNcIdBytesReceiveMap);
 209             pmagent.connectToPMAgent(ncIdOFPortDurationMap, ncIdOFPortReceiveDropMap, ncIdOFPortReceiveError, ncIdPacketSentMap, ncIdPacketReceiveMap, ncIdBytesSentMap, ncIdBytesReceiveMap);
 210         }
 211
 212         /*
 213          * Input allNodesStats contains statistics of all nodeConnectors of all nodes. Key is the node and values contains another map with key as node connector and value as statresult.
 214          * Output will be a map with key as nodeconnector and value as the statresult. The key contains nodeconnectors of all the nodes.
 215          */
 216     }
 217
 218     private Map<String, String> combineAllNodesStats(Map<String, Map<String, String>> allNodesStats) {
 219         Map<String, String> allNcsStatsMap = new HashMap<String, String>();
 220         for (Map.Entry<String, Map<String, String>> entry : allNodesStats.entrySet()) {
 221             Map<String, String> ncStatsMap = entry.getValue();
 222             for (Map.Entry<String, String> statResult : ncStatsMap.entrySet()) {
 223                 allNcsStatsMap.put(statResult.getKey(), statResult.getValue());
 224             }
 225         }
 226         return allNcsStatsMap;
 227     }
 228
 229     /*
 230      * FlowRpcStatisticsListener listens for the FlowTableStatisticsUpdate and then update the corresponding counter map
 231      */
 232     class FlowRpcStatisticsListener implements OpendaylightFlowTableStatisticsListener {
 233
 234         @Override
 235         public void onFlowTableStatisticsUpdate(FlowTableStatisticsUpdate flowTableStats) {
 236             String node = flowTableStats.getId().getValue().split(":")[1];
 237             Map<String, String> entriesPerOFTableMap = new HashMap<String, String>();
 238             List<FlowTableAndStatisticsMap> flowTableAndStatisticsMapList = flowTableStats.getFlowTableAndStatisticsMap();
 239             for (FlowTableAndStatisticsMap flowTableAndStatisticsMap : flowTableAndStatisticsMapList) {
 240                 String nodeTableStr =  "dpnId_" + node + "_table_" + flowTableAndStatisticsMap.getTableId().getValue().toString();
 241                 entriesPerOFTableMap.put("EntriesPerOFTable:" + nodeTableStr + "_EntriesPerOFTable", flowTableAndStatisticsMap.getActiveFlows().getValue().toString());
 242             }
 243             nodeAndEntriesPerOFTableMap.put(node, entriesPerOFTableMap);
 244             entriesPerOFTableMap = combineAllNodesStats(nodeAndEntriesPerOFTableMap);
 245             pmagent.connectToPMAgentAndInvokeEntriesPerOFTable(entriesPerOFTableMap);
 246         }
 247
 248     }
 249
 250     @Override
 251     protected void remove(InstanceIdentifier<Node> identifier, Node node) {
 252         NodeId nodeId = node.getId();
 253         String nodeVal = nodeId.getValue().split(":")[1];
 254         BigInteger dpId = new BigInteger(nodeVal);
 255         if (nodes.contains(dpId)) {
 256             nodes.remove(dpId);
 257             nodeAndNcIdOFPortDurationMap.remove(nodeVal);
 258             nodeAndNcIdOFPortReceiveDropMap.remove(nodeVal);
 259             nodeAndNcIdOFPortReceiveError.remove(nodeVal);
 260             nodeAndNcIdPacketSentMap.remove(nodeVal);
 261             nodeAndNcIdPacketReceiveMap.remove(nodeVal);
 262             nodeAndNcIdBytesSentMap.remove(nodeVal);
 263             nodeAndNcIdBytesReceiveMap.remove(nodeVal);
 264             nodeAndEntriesPerOFTableMap.remove(nodeVal);
 265         }
 266         if (nodes.isEmpty()) {
 267             stopPortStatRequestTask();
 268         }
 269     }
 270
 271     @Override
 272     protected void update(InstanceIdentifier<Node> identifier, Node original,
 273             Node update) {
 274         // TODO Auto-generated method stub
 275     }
 276
 277     @Override
 278     protected void add(InstanceIdentifier<Node> identifier, Node node) {
 279         NodeId nodeId = node.getId();
 280         BigInteger dpId = new BigInteger(nodeId.getValue().split(":")[1]);
 281         if (nodes.contains(dpId)) {
 282             return;
 283         }
 284         nodes.add(dpId);
 285         if (nodes.size() == 1) {
 286             schedulePortStatRequestTask();
 287         }
 288     }
 289 }