Bug 781: Statistics manager is throwing IllegalStateException
[controller.git] / opendaylight / md-sal / statistics-manager / src / main / java / org / opendaylight / controller / md / statistics / manager / NodeStatisticsHandler.java
index 45788b331954ceaad947977d8cc61ec9e4e4f25d..5ace260251673bb9fa42b9da14a95a5ece724b4c 100644 (file)
@@ -9,9 +9,10 @@ package org.opendaylight.controller.md.statistics.manager;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.TimeUnit;
 
-import org.opendaylight.controller.md.statistics.manager.MultipartMessageManager.StatsRequestType;
 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
@@ -20,13 +21,16 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.ta
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsData;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsDataBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.aggregate.flow.statistics.AggregateFlowStatisticsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.flow.and.statistics.map.list.FlowAndStatisticsMapList;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.flow.table.and.statistics.map.FlowTableAndStatisticsMap;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionAware;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeatures;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.NodeGroupFeaturesBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.group.features.GroupFeaturesBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupFeatures;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.desc.stats.reply.GroupDescStats;
@@ -38,12 +42,15 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.N
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeatures;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.NodeMeterFeaturesBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.nodes.node.MeterFeaturesBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterFeatures;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.config.stats.reply.MeterConfigStats;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.meter.statistics.reply.MeterStats;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.AggregateFlowStatistics;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.queue.id.and.statistics.map.QueueIdAndStatisticsMap;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 import org.slf4j.Logger;
@@ -58,11 +65,14 @@ import com.google.common.base.Preconditions;
  *
  * @author avishnoi@in.ibm.com
  */
-public final class NodeStatisticsHandler implements AutoCloseable {
+public final class NodeStatisticsHandler implements AutoCloseable, FlowCapableContext {
     private static final Logger logger = LoggerFactory.getLogger(NodeStatisticsHandler.class);
+
+    private static final long STATS_COLLECTION_MILLIS = TimeUnit.SECONDS.toMillis(15);
+    private static final long FIRST_COLLECTION_MILLIS = TimeUnit.SECONDS.toMillis(5);
     private static final int NUMBER_OF_WAIT_CYCLES = 2;
 
-    private final MultipartMessageManager msgManager = new MultipartMessageManager();
+    private final MultipartMessageManager msgManager;
     private final InstanceIdentifier<Node> targetNodeIdentifier;
     private final FlowStatsTracker flowStats;
     private final FlowTableStatsTracker flowTableStats;
@@ -75,84 +85,106 @@ public final class NodeStatisticsHandler implements AutoCloseable {
     private final DataProviderService dps;
     private final NodeRef targetNodeRef;
     private final NodeKey targetNodeKey;
-
-    public NodeStatisticsHandler(final DataProviderService dps, final NodeKey nodeKey) {
+    private final TimerTask task = new TimerTask() {
+        @Override
+        public void run() {
+            try{
+                requestPeriodicStatistics();
+                cleanStaleStatistics();
+            }catch(Exception e){
+                logger.warn("Exception occured while sending statistics request : {}",e);
+            }
+        }
+    };
+
+    public NodeStatisticsHandler(final DataProviderService dps, final NodeKey nodeKey,
+            final OpendaylightFlowStatisticsService flowStatsService,
+            final OpendaylightFlowTableStatisticsService flowTableStatsService,
+            final OpendaylightGroupStatisticsService groupStatsService,
+            final OpendaylightMeterStatisticsService meterStatsService,
+            final OpendaylightPortStatisticsService portStatsService,
+            final OpendaylightQueueStatisticsService queueStatsService) {
         this.dps = Preconditions.checkNotNull(dps);
         this.targetNodeKey = Preconditions.checkNotNull(nodeKey);
         this.targetNodeIdentifier = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey).build();
         this.targetNodeRef = new NodeRef(targetNodeIdentifier);
 
-        final long lifetimeNanos = TimeUnit.MILLISECONDS.toNanos(StatisticsProvider.STATS_COLLECTION_MILLIS * NUMBER_OF_WAIT_CYCLES);
-        flowStats = new FlowStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        flowTableStats = new FlowTableStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        groupDescStats = new GroupDescStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        groupStats = new GroupStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        meterConfigStats = new MeterConfigStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        meterStats = new MeterStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        nodeConnectorStats = new NodeConnectorStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
-        queueStats = new QueueStatsTracker(targetNodeIdentifier, dps, lifetimeNanos);
+        final long lifetimeNanos = TimeUnit.MILLISECONDS.toNanos(STATS_COLLECTION_MILLIS * NUMBER_OF_WAIT_CYCLES);
+
+        msgManager = new MultipartMessageManager(lifetimeNanos);
+        flowStats = new FlowStatsTracker(flowStatsService, this, lifetimeNanos);
+        flowTableStats = new FlowTableStatsTracker(flowTableStatsService, this, lifetimeNanos);
+        groupDescStats = new GroupDescStatsTracker(groupStatsService, this, lifetimeNanos);
+        groupStats = new GroupStatsTracker(groupStatsService, this, lifetimeNanos);
+        meterConfigStats = new MeterConfigStatsTracker(meterStatsService, this, lifetimeNanos);
+        meterStats = new MeterStatsTracker(meterStatsService, this, lifetimeNanos);
+        nodeConnectorStats = new NodeConnectorStatsTracker(portStatsService, this, lifetimeNanos);
+        queueStats = new QueueStatsTracker(queueStatsService, this, lifetimeNanos);
     }
 
     public NodeKey getTargetNodeKey() {
         return targetNodeKey;
     }
 
-    public Collection<TableKey> getKnownTables() {
-        return flowTableStats.getTables();
-    }
-
-    public InstanceIdentifier<Node> getTargetNodeIdentifier() {
+    @Override
+    public InstanceIdentifier<Node> getNodeIdentifier() {
         return targetNodeIdentifier;
     }
 
-    public NodeRef getTargetNodeRef() {
+    @Override
+    public NodeRef getNodeRef() {
         return targetNodeRef;
     }
 
-    public synchronized void updateGroupDescStats(TransactionAware transaction, Boolean more, List<GroupDescStats> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    @Override
+    public DataModificationTransaction startDataModification() {
+        return dps.beginTransaction();
+    }
+
+    public synchronized void updateGroupDescStats(TransactionAware transaction, List<GroupDescStats> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             groupDescStats.updateStats(list);
         }
     }
 
-    public synchronized void updateGroupStats(TransactionAware transaction, Boolean more, List<GroupStats> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateGroupStats(TransactionAware transaction, List<GroupStats> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             groupStats.updateStats(list);
         }
     }
 
-    public synchronized void updateMeterConfigStats(TransactionAware transaction, Boolean more, List<MeterConfigStats> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateMeterConfigStats(TransactionAware transaction, List<MeterConfigStats> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             meterConfigStats.updateStats(list);
         }
     }
 
-    public synchronized void updateMeterStats(TransactionAware transaction, Boolean more, List<MeterStats> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateMeterStats(TransactionAware transaction, List<MeterStats> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             meterStats.updateStats(list);
         }
     }
 
-    public synchronized void updateQueueStats(TransactionAware transaction, Boolean more, List<QueueIdAndStatisticsMap> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateQueueStats(TransactionAware transaction, List<QueueIdAndStatisticsMap> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             queueStats.updateStats(list);
         }
     }
 
-    public synchronized void updateFlowTableStats(TransactionAware transaction, Boolean more, List<FlowTableAndStatisticsMap> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateFlowTableStats(TransactionAware transaction, List<FlowTableAndStatisticsMap> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             flowTableStats.updateStats(list);
         }
     }
 
-    public synchronized void updateNodeConnectorStats(TransactionAware transaction, Boolean more, List<NodeConnectorStatisticsAndPortNumberMap> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateNodeConnectorStats(TransactionAware transaction, List<NodeConnectorStatisticsAndPortNumberMap> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             nodeConnectorStats.updateStats(list);
         }
     }
 
-    public synchronized void updateAggregateFlowStats(TransactionAware transaction, Boolean more, AggregateFlowStatistics flowStats) {
-        final Short tableId = msgManager.isExpectedTableTransaction(transaction, more);
+    public synchronized void updateAggregateFlowStats(TransactionAware transaction, AggregateFlowStatistics flowStats) {
+        final Short tableId = msgManager.isExpectedTableTransaction(transaction);
         if (tableId != null) {
             final DataModificationTransaction trans = dps.beginTransaction();
             InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class).child(Node.class, targetNodeKey)
@@ -175,8 +207,8 @@ public final class NodeStatisticsHandler implements AutoCloseable {
         }
     }
 
-    public synchronized void updateFlowStats(TransactionAware transaction, Boolean more, List<FlowAndStatisticsMapList> list) {
-        if (msgManager.isExpectedTransaction(transaction, more)) {
+    public synchronized void updateFlowStats(TransactionAware transaction, List<FlowAndStatisticsMapList> list) {
+        if (msgManager.isExpectedTransaction(transaction)) {
             flowStats.updateStats(list);
         }
     }
@@ -233,19 +265,66 @@ public final class NodeStatisticsHandler implements AutoCloseable {
         trans.commit();
     }
 
+    public synchronized void requestPeriodicStatistics() {
+        logger.debug("Send requests for statistics collection to node : {}", targetNodeKey);
+
+        flowTableStats.request();
+
+        // FIXME: it does not make sense to trigger this before sendAllFlowTablesStatisticsRequest()
+        //        comes back -- we do not have any tables anyway.
+        final Collection<TableKey> tables = flowTableStats.getTables();
+        logger.debug("Node {} supports {} table(s)", targetNodeKey, tables.size());
+        for (final TableKey key : tables) {
+            logger.debug("Send aggregate stats request for flow table {} to node {}", key.getId(), targetNodeKey);
+            flowStats.requestAggregateFlows(key);
+        }
+
+        flowStats.requestAllFlowsAllTables();
+        nodeConnectorStats.request();
+        groupStats.request();
+        groupDescStats.request();
+        meterStats.request();
+        meterConfigStats.request();
+        queueStats.request();
+    }
+
+    public synchronized void start(final Timer timer) {
+        flowStats.start(dps);
+        groupDescStats.start(dps);
+        groupStats.start(dps);
+        meterConfigStats.start(dps);
+        meterStats.start(dps);
+        queueStats.start(dps);
+
+        timer.schedule(task, (long) (Math.random() * FIRST_COLLECTION_MILLIS), STATS_COLLECTION_MILLIS);
+
+        logger.debug("Statistics handler for node started with base interval {}ms", STATS_COLLECTION_MILLIS);
+
+        requestPeriodicStatistics();
+    }
+
     @Override
-    public void close() {
-        // FIXME: cleanup any resources we hold (registrations, etc.)
+    public synchronized void close() {
+        task.cancel();
+        flowStats.close();
+        groupDescStats.close();
+        groupStats.close();
+        meterConfigStats.close();
+        meterStats.close();
+        queueStats.close();
+
         logger.debug("Statistics handler for {} shut down", targetNodeKey.getId());
     }
 
-    // FIXME: this should be private
-    public synchronized void recordExpectedTransaction(TransactionId transactionId, StatsRequestType reqType) {
-        msgManager.recordExpectedTransaction(transactionId, reqType);
+    @Override
+    public void registerTransaction(TransactionId id) {
+        msgManager.recordExpectedTransaction(id);
+        logger.debug("Transaction {} for node {} sent successfully", id, targetNodeKey);
     }
 
-    // FIXME: this should be private
-    public synchronized void recordExpectedTableTransaction(TransactionId transactionId, Short tableId) {
-        msgManager.recordExpectedTableTransaction(transactionId, StatsRequestType.AGGR_FLOW, tableId);
+    @Override
+    public void registerTableTransaction(final TransactionId id, final Short table) {
+        msgManager.recordExpectedTableTransaction(id, table);
+        logger.debug("Transaction {} for node {} table {} sent successfully", id, targetNodeKey, table);
     }
 }