Adding multi-instance support for flow stats.

Change-Id: I428c5a7cb58f4f9773a125fc94fb368ed846cb0d

@@ -61,6 +61,22 @@ public final class DefaultPortStatistics implements PortStatistics {
         this.durationNano = durationNano;
     }
 
+    // Constructor for serializer
+    private DefaultPortStatistics() {
+        this.deviceId = null;
+        this.port = 0;
+        this.packetsReceived = 0;
+        this.packetsSent = 0;
+        this.bytesReceived = 0;
+        this.bytesSent = 0;
+        this.packetsRxDropped = 0;
+        this.packetsTxDropped = 0;
+        this.packetsRxErrors = 0;
+        this.packetsTxErrors = 0;
+        this.durationSec = 0;
+        this.durationNano = 0;
+    }
+
     /**
      * Creates a builder for DefaultPortStatistics object.
      *

@@ -50,6 +50,7 @@ import org.onosproject.net.OduCltPort;
 import org.onosproject.net.OmsPort;
 import org.onosproject.net.Port;
 import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DefaultPortStatistics;
 import org.onosproject.net.device.DeviceClockService;
 import org.onosproject.net.device.DeviceDescription;
 import org.onosproject.net.device.DeviceEvent;
@@ -68,8 +69,16 @@ import org.onosproject.store.cluster.messaging.ClusterMessage;
 import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
 import org.onosproject.store.cluster.messaging.MessageSubject;
 import org.onosproject.store.impl.Timestamped;
+import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoSerializer;
 import org.onosproject.store.serializers.custom.DistributedStoreSerializers;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapEvent;
+import org.onosproject.store.service.EventuallyConsistentMapListener;
+import org.onosproject.store.service.MultiValuedTimestamp;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.WallClockTimestamp;
+import org.onosproject.store.service.WallclockClockManager;
 import org.slf4j.Logger;
 
 import java.io.IOException;
@@ -102,6 +111,7 @@ import static org.onosproject.net.DefaultAnnotations.merge;
 import static org.onosproject.net.device.DeviceEvent.Type.*;
 import static org.onosproject.net.device.DeviceEvent.Type.DEVICE_REMOVED;
 import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.*;
+import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
@@ -128,8 +138,10 @@ public class GossipDeviceStore
     // cache of Device and Ports generated by compositing descriptions from providers
     private final ConcurrentMap<DeviceId, Device> devices = Maps.newConcurrentMap();
     private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = Maps.newConcurrentMap();
-    private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, PortStatistics>>
-            devicePortStats = Maps.newConcurrentMap();
+
+    private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortStats;
+    private final EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>>
+            portStatsListener = new InternalPortStatsListener();
 
     // to be updated under Device lock
     private final Map<DeviceId, Timestamp> offline = Maps.newHashMap();
@@ -142,6 +154,9 @@ public class GossipDeviceStore
     protected DeviceClockService deviceClockService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ClusterCommunicationService clusterCommunicator;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
@@ -182,10 +197,8 @@ public class GossipDeviceStore
     private long initialDelaySec = 5;
     private long periodSec = 5;
 
-
     @Activate
     public void activate() {
-
         executor = Executors.newCachedThreadPool(groupedThreads("onos/device", "fg-%d"));
 
         backgroundExecutor =
@@ -219,6 +232,22 @@ public class GossipDeviceStore
         backgroundExecutor.scheduleAtFixedRate(new SendAdvertisementTask(),
                                                initialDelaySec, periodSec, TimeUnit.SECONDS);
 
+        // Create a distributed map for port stats.
+        KryoNamespace.Builder deviceDataSerializer = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.API)
+                .register(DefaultPortStatistics.class)
+                .register(DeviceId.class)
+                .register(MultiValuedTimestamp.class)
+                .register(WallClockTimestamp.class);
+
+        devicePortStats = storageService.<DeviceId, Map<PortNumber, PortStatistics>>eventuallyConsistentMapBuilder()
+                .withName("port-stats")
+                .withSerializer(deviceDataSerializer)
+                .withAntiEntropyPeriod(5, TimeUnit.SECONDS)
+                .withClockService(new WallclockClockManager<>())
+                .withTombstonesDisabled()
+                .build();
+        devicePortStats.addListener(portStatsListener);
         log.info("Started");
     }
 
@@ -702,7 +731,6 @@ public class GossipDeviceStore
     private DeviceDescriptions getOrCreateProviderDeviceDescriptions(
             Map<ProviderId, DeviceDescriptions> device,
             ProviderId providerId, Timestamped<DeviceDescription> deltaDesc) {
-
         synchronized (device) {
             DeviceDescriptions r = device.get(providerId);
             if (r == null) {
@@ -747,7 +775,6 @@ public class GossipDeviceStore
 
     private DeviceEvent updatePortStatusInternal(ProviderId providerId, DeviceId deviceId,
                                                  Timestamped<PortDescription> deltaDesc) {
-
         Device device = devices.get(deviceId);
         checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
 
@@ -805,24 +832,21 @@ public class GossipDeviceStore
     @Override
     public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
                                             Collection<PortStatistics> portStats) {
-
-        ConcurrentMap<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId);
+        Map<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId);
         if (statsMap == null) {
-            statsMap = Maps.newConcurrentMap();
-            devicePortStats.put(deviceId, statsMap);
+            statsMap = Maps.newHashMap();
         }
 
-        for (PortStatistics stat: portStats) {
+        for (PortStatistics stat : portStats) {
             PortNumber portNumber = PortNumber.portNumber(stat.port());
             statsMap.put(portNumber, stat);
         }
-
-        return new DeviceEvent(PORT_STATS_UPDATED,  devices.get(deviceId), null);
+        devicePortStats.put(deviceId, statsMap);
+        return null; // new DeviceEvent(PORT_STATS_UPDATED,  devices.get(deviceId), null);
     }
 
     @Override
     public List<PortStatistics> getPortStatistics(DeviceId deviceId) {
-
         Map<PortNumber, PortStatistics> portStats = devicePortStats.get(deviceId);
         if (portStats == null) {
             return Collections.emptyList();
@@ -978,7 +1002,7 @@ public class GossipDeviceStore
             annotations = merge(annotations, e.getValue().getDeviceDesc().value().annotations());
         }
 
-        return new DefaultDevice(primary, deviceId , type, manufacturer,
+        return new DefaultDevice(primary, deviceId, type, manufacturer,
                                  hwVersion, swVersion, serialNumber,
                                  chassisId, annotations);
     }
@@ -1192,7 +1216,7 @@ public class GossipDeviceStore
 
     /**
      * Responds to anti-entropy advertisement message.
-     * <P>
+     * <p/>
      * Notify sender about out-dated information using regular replication message.
      * Send back advertisement to sender if not in sync.
      *
@@ -1386,7 +1410,6 @@ public class GossipDeviceStore
             implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
-
             log.debug("Received device update event from peer: {}", message.sender());
             InternalDeviceEvent event = SERIALIZER.decode(message.payload());
 
@@ -1406,7 +1429,6 @@ public class GossipDeviceStore
             implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
-
             log.debug("Received device offline event from peer: {}", message.sender());
             InternalDeviceOfflineEvent event = SERIALIZER.decode(message.payload());
 
@@ -1440,7 +1462,6 @@ public class GossipDeviceStore
             implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
-
             log.debug("Received device removed event from peer: {}", message.sender());
             InternalDeviceRemovedEvent event = SERIALIZER.decode(message.payload());
 
@@ -1509,7 +1530,6 @@ public class GossipDeviceStore
 
     private final class InternalDeviceAdvertisementListener
             implements ClusterMessageHandler {
-
         @Override
         public void handle(ClusterMessage message) {
             log.trace("Received Device Anti-Entropy advertisement from peer: {}", message.sender());
@@ -1526,7 +1546,6 @@ public class GossipDeviceStore
             implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
-
             log.debug("Received injected device event from peer: {}", message.sender());
             DeviceInjectedEvent event = SERIALIZER.decode(message.payload());
 
@@ -1551,7 +1570,6 @@ public class GossipDeviceStore
             implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
-
             log.debug("Received injected port event from peer: {}", message.sender());
             PortInjectedEvent event = SERIALIZER.decode(message.payload());
 
@@ -1571,4 +1589,17 @@ public class GossipDeviceStore
             }
         }
     }
+
+    private class InternalPortStatsListener
+            implements EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>> {
+        @Override
+        public void event(EventuallyConsistentMapEvent<DeviceId, Map<PortNumber, PortStatistics>> event) {
+            if (event.type() == PUT) {
+                Device device = devices.get(event.key());
+                if (device != null) {
+                    delegate.notify(new DeviceEvent(PORT_STATS_UPDATED, device));
+                }
+            }
+        }
+    }
 }

@@ -55,6 +55,7 @@ import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
 import org.onosproject.store.cluster.messaging.ClusterMessage;
 import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
 import org.onosproject.store.cluster.messaging.MessageSubject;
+import org.onosproject.store.consistent.impl.DatabaseManager;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -165,6 +166,10 @@ public class GossipDeviceStoreTest {
         testGossipDeviceStore = new TestGossipDeviceStore(deviceClockService, clusterService, clusterCommunicator);
         testGossipDeviceStore.mastershipService = new TestMastershipService();
 
+        TestDatabaseManager testDatabaseManager = new TestDatabaseManager();
+        testDatabaseManager.init(clusterService, clusterCommunicator);
+        testGossipDeviceStore.storageService = testDatabaseManager;
+
         gossipDeviceStore = testGossipDeviceStore;
         gossipDeviceStore.activate();
         deviceStore = gossipDeviceStore;
@@ -885,4 +890,12 @@ public class GossipDeviceStoreTest {
             nodeStates.put(NID2, ACTIVE);
         }
     }
+
+    private class TestDatabaseManager extends DatabaseManager {
+        void init(ClusterService clusterService,
+                  ClusterCommunicationService clusterCommunicator) {
+            this.clusterService = clusterService;
+            this.clusterCommunicator = clusterCommunicator;
+        }
+    }
 }

@@ -71,9 +71,11 @@ import org.onosproject.net.Port;
 import org.onosproject.net.PortNumber;
 import org.onosproject.net.device.DefaultDeviceDescription;
 import org.onosproject.net.device.DefaultPortDescription;
+import org.onosproject.net.device.DefaultPortStatistics;
 import org.onosproject.net.device.OchPortDescription;
 import org.onosproject.net.device.OduCltPortDescription;
 import org.onosproject.net.device.OmsPortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.flow.CompletedBatchOperation;
 import org.onosproject.net.flow.DefaultFlowEntry;
 import org.onosproject.net.flow.DefaultFlowRule;
@@ -380,7 +382,9 @@ public final class KryoNamespaces {
                     IntentOperation.class,
                     FlowRuleExtPayLoad.class,
                     Frequency.class,
-                    DefaultAnnotations.class
+                    DefaultAnnotations.class,
+                    PortStatistics.class,
+                    DefaultPortStatistics.class
             )
             .register(new DefaultApplicationIdSerializer(), DefaultApplicationId.class)
             .register(new URISerializer(), URI.class)

@@ -14,7 +14,7 @@ cut -c7- $aux | cut -d\  -f1 | sort > $aux.1
 
 # Normalize the expected apps
 apps=${ONOS_APPS:-drivers,openflow}
-(for app in ${apps/,/ }; do echo org.onosproject.$app; done) | sort > $aux.2
+(for app in ${apps//,/ }; do echo org.onosproject.$app; done) | sort > $aux.2
 
 # Check for differences
 diff $aux.1 $aux.2

+# Madan's ProxMox ONOS instances 1,2,3 & ONOS mininet box
+
+export ONOS_NIC="10.128.4.*"
+export OC1="10.128.4.2"
+export OC2="10.128.4.3"
+export OC3="10.128.4.4"
+export OCN="10.128.4.5"

-# Office ProxMox ONOS instances 1,2,3 & ONOS mininet box
+# Tom's ProxMox ONOS instances 1,2,3 & ONOS mininet box
 
-export ONOS_NIC=10.128.11.*
+export ONOS_NIC="10.128.11.*"
 export OC1="10.128.11.1"
 export OC2="10.128.11.2"
 export OC3="10.128.11.3"

@@ -850,7 +850,7 @@ public abstract class TopologyViewMessageHandlerBase extends UiMessageHandler {
             if (load != null) {
                 this.hasTraffic = hasTraffic || load.rate() > threshold;
                 this.bytes += load.latest();
-                this.rate = load.rate();
+                this.rate += load.rate();
             }
         }