ONOS-1440: Implements port statistics feature, which polls port statistics of al…

…l devices every 10 seconds. Also, implemented a simple portstats ONOS CLI command to show the statistics.

Change-Id: I57e046ae2c2463a58b478d3a5b523422cde71ba2

+package org.onosproject.cli.net;
+
+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.karaf.shell.commands.Command;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.device.PortStatistics;
+
+/**
+ * Lists port statistic of all ports in the system.
+ */
+@Command(scope = "onos", name = "portstats",
+        description = "Lists statistics of all ports in the system")
+public class DevicePortStatsCommand extends AbstractShellCommand {
+
+    private static final String FORMAT =
+            "   port=%s, pktRx=%s, pktTx=%s, bytesRx=%s, bytesTx=%s, pktRxDrp=%s, pktTxDrp=%s, Dur=%s";
+
+    @Override
+    protected void execute() {
+        DeviceService deviceService = get(DeviceService.class);
+
+        deviceService.getDevices().forEach(d ->
+                        printPortStats(d.id(), deviceService.getPortStatistics(d.id()))
+        );
+    }
+
+    private void printPortStats(DeviceId deviceId, Iterable<PortStatistics> portStats) {
+        print("deviceId=%s", deviceId);
+        for (PortStatistics stat : portStats) {
+            print(FORMAT, stat.port(), stat.packetsReceived(), stat.packetsSent(), stat.bytesReceived(),
+                    stat.bytesSent(), stat.packetsRxDropped(), stat.packetsTxDropped(), stat.durationSec());
+        }
+    }
+}

@@ -280,6 +280,10 @@
         </command>
 
         <command>
+            <action class="org.onosproject.cli.net.DevicePortStatsCommand"/>
+        </command>
+
+        <command>
             <action class="org.onosproject.cli.net.FlowsListCommand"/>
             <completers>
                 <ref component-id="flowRuleStatusCompleter"/>

+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.net.device;
+
+import org.onosproject.net.DeviceId;
+
+/**
+ * Default implementation of immutable port statistics.
+ */
+public final class DefaultPortStatistics implements PortStatistics {
+
+    private final DeviceId deviceId;
+    private final int  port;
+    private final long packetsReceived;
+    private final long packetsSent;
+    private final long bytesReceived;
+    private final long bytesSent;
+    private final long packetsRxDropped;
+    private final long packetsTxDropped;
+    private final long packetsRxErrors;
+    private final long packetsTxErrors;
+    private final long durationSec;
+    private final long durationNano;
+
+    private DefaultPortStatistics(DeviceId deviceId,
+                                  int  port,
+                                  long packetsReceived,
+                                  long packetsSent,
+                                  long bytesReceived,
+                                  long bytesSent,
+                                  long packetsRxDropped,
+                                  long packetsTxDropped,
+                                  long packetsRxErrors,
+                                  long packetsTxErrors,
+                                  long durationSec,
+                                  long durationNano) {
+        this.deviceId = deviceId;
+        this.port = port;
+        this.packetsReceived = packetsReceived;
+        this.packetsSent = packetsSent;
+        this.bytesReceived = bytesReceived;
+        this.bytesSent = bytesSent;
+        this.packetsRxDropped = packetsRxDropped;
+        this.packetsTxDropped = packetsTxDropped;
+        this.packetsRxErrors = packetsRxErrors;
+        this.packetsTxErrors = packetsTxErrors;
+        this.durationSec = durationSec;
+        this.durationNano = durationNano;
+    }
+
+    /**
+     * Creates a builder for DefaultPortStatistics object.
+     *
+     * @return builder object for DefaultPortStatistics object
+     */
+    public static DefaultPortStatistics.Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public int port() {
+        return this.port;
+    }
+
+    @Override
+    public long packetsReceived() {
+        return this.packetsReceived;
+    }
+
+    @Override
+    public long packetsSent() {
+        return this.packetsSent;
+    }
+
+    @Override
+    public long bytesReceived() {
+        return this.bytesReceived;
+    }
+
+    @Override
+    public long bytesSent() {
+        return this.bytesSent;
+    }
+
+    @Override
+    public long packetsRxDropped() {
+        return this.packetsRxDropped;
+    }
+
+    @Override
+    public long packetsTxDropped() {
+        return this.packetsTxDropped;
+    }
+
+    @Override
+    public long packetsRxErrors() {
+        return this.packetsRxErrors;
+    }
+
+    @Override
+    public long packetsTxErrors() {
+        return this.packetsTxErrors;
+    }
+
+    @Override
+    public long durationSec() {
+        return this.durationSec;
+    }
+
+    @Override
+    public long durationNano() {
+        return this.durationNano;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder("device: " + deviceId + ", ");
+
+        sb.append("port: " + this.port + ", ");
+        sb.append("pktRx: " + this.packetsReceived + ", ");
+        sb.append("pktTx: " + this.packetsSent + ", ");
+        sb.append("byteRx: " + this.bytesReceived + ", ");
+        sb.append("byteTx: " + this.bytesSent + ", ");
+        sb.append("pktRxErr: " + this.packetsRxErrors + ", ");
+        sb.append("pktTxErr: " + this.packetsTxErrors + ", ");
+        sb.append("pktRxDrp: " + this.packetsRxDropped + ", ");
+        sb.append("pktTxDrp: " + this.packetsTxDropped);
+
+        return sb.toString();
+    }
+
+    public static final class Builder {
+
+        DeviceId deviceId;
+        int port;
+        long packetsReceived;
+        long packetsSent;
+        long bytesReceived;
+        long bytesSent;
+        long packetsRxDropped;
+        long packetsTxDropped;
+        long packetsRxErrors;
+        long packetsTxErrors;
+        long durationSec;
+        long durationNano;
+
+        private Builder() {
+
+        }
+
+        /**
+         * Sets port number.
+         *
+         * @param port port number
+         * @return builder object
+         */
+        public Builder setPort(int port) {
+            this.port = port;
+
+            return this;
+        }
+
+        /**
+         * Sets the device identifier.
+         *
+         * @param deviceId device identifier
+         * @return builder object
+         */
+        public Builder setDeviceId(DeviceId deviceId) {
+            this.deviceId = deviceId;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of packet received.
+         *
+         * @param packets number of packets received
+         * @return  builder object
+         */
+        public Builder setPacketsReceived(long packets) {
+            packetsReceived = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of packets sent.
+         *
+         * @param packets number of packets sent
+         * @return  builder object
+         */
+        public Builder setPacketsSent(long packets) {
+            packetsSent = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of received bytes.
+         *
+         * @param bytes number of received bytes.
+         * @return  builder object
+         */
+        public Builder setBytesReceived(long bytes) {
+            bytesReceived = bytes;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of sent bytes.
+         *
+         * @param bytes number of sent bytes
+         * @return builder object
+         */
+        public Builder setBytesSent(long bytes) {
+            bytesSent = bytes;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of packets dropped by RX.
+         *
+         * @param packets  number of packets dropped by RX
+         * @return builder object
+         */
+        public Builder setPacketsRxDropped(long packets) {
+            packetsRxDropped = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of packets dropped by TX.
+         *
+         * @param packets
+         * @return builder object
+         */
+        public Builder setPacketsTxDropped(long packets) {
+            packetsTxDropped = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of receive errors.
+         *
+         * @param packets number of receive errors
+         * @return builder object
+         */
+        public Builder setPacketsRxErrors(long packets) {
+            packetsRxErrors = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the number of transmit errors.
+         *
+         * @param packets number of transmit errors
+         * @return builder object
+         */
+        public Builder setPacketsTxErrors(long packets) {
+            packetsTxErrors = packets;
+
+            return this;
+        }
+
+        /**
+         * Sets the time port has been alive in seconds.
+         *
+         * @param sec time port has been alive in seconds
+         * @return builder object
+         */
+        public Builder setDurationSec(long sec) {
+            durationSec = sec;
+
+            return this;
+        }
+
+        /**
+         * Sets the time port has been alive in nano seconds.
+         *
+         * @param nano time port has been alive in nano seconds
+         * @return builder object
+         */
+        public Builder setDurationNano(long nano) {
+            durationNano = nano;
+
+            return this;
+        }
+
+        /**
+         * Creates a PortStatistics object.
+         *
+         * @return DefaultPortStatistics object
+         */
+        public DefaultPortStatistics build() {
+            return new DefaultPortStatistics(
+                    deviceId,
+                    port,
+                    packetsReceived,
+                    packetsSent,
+                    bytesReceived,
+                    bytesSent,
+                    packetsRxDropped,
+                    packetsTxDropped,
+                    packetsRxErrors,
+                    packetsTxErrors,
+                    durationSec,
+                    durationNano);
+        }
+
+    }
+}

@@ -72,7 +72,12 @@ public class DeviceEvent extends AbstractEvent<DeviceEvent.Type, Device> {
         /**
          * Signifies that a port has been removed.
          */
-        PORT_REMOVED
+        PORT_REMOVED,
+
+        /*
+         * Signifies that port statistics has been updated.
+         */
+        PORT_STATS_UPDATED
     }
 
     /**

@@ -19,6 +19,7 @@ import org.onosproject.net.DeviceId;
 import org.onosproject.net.MastershipRole;
 import org.onosproject.net.provider.ProviderService;
 
+import java.util.Collection;
 import java.util.List;
 
 /**
@@ -70,4 +71,12 @@ public interface DeviceProviderService extends ProviderService<DeviceProvider> {
      */
     void receivedRoleReply(DeviceId deviceId, MastershipRole requested, MastershipRole response);
 
+    /**
+     * Sends statistics about all ports of a device.
+     *
+     * @param deviceId          identity of the device
+     * @param portStatistics  list of device port statistics
+     */
+    void updatePortStatistics(DeviceId deviceId, Collection<PortStatistics> portStatistics);
+
 }

@@ -78,6 +78,14 @@ public interface DeviceService {
     List<Port> getPorts(DeviceId deviceId);
 
     /**
+     * Returns the list of port statistics associated with the device.
+     *
+     * @param deviceId device identitifer
+     * @return list of port statistics
+     */
+    List<PortStatistics> getPortStatistics(DeviceId deviceId);
+
+    /**
      * Returns the port with the specified number and hosted by the given device.
      *
      * @param deviceId   device identifier

@@ -22,6 +22,7 @@ import org.onosproject.net.PortNumber;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.store.Store;
 
+import java.util.Collection;
 import java.util.List;
 
 /**
@@ -114,6 +115,26 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> {
     List<Port> getPorts(DeviceId deviceId);
 
     /**
+     * Updates the port statistics of the specified device using the give port
+     * statistics.
+     *
+     * @param providerId  provider identifier
+     * @param deviceId    device identifier
+     * @param portStats   list of port statistics
+     * @return ready to send event describing what occurred;
+     */
+    DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
+                                     Collection<PortStatistics> portStats);
+
+    /**
+     * Returns the list of port statistics of the specified device.
+     *
+     * @param deviceId device identifier
+     * @return list of port statistics of all ports of the device
+     */
+    List<PortStatistics> getPortStatistics(DeviceId deviceId);
+
+    /**
      * Returns the specified device port.
      *
      * @param deviceId   device identifier
@@ -137,4 +158,6 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> {
      * @return null if no such device, or was forwarded to remove master
      */
     DeviceEvent removeDevice(DeviceId deviceId);
+
+
 }

+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.net.device;
+
+/**
+ * Statistics of a port.
+ */
+public interface PortStatistics {
+
+    /**
+     * Returns the port number.
+     *
+     * @return port number
+     */
+    public int  port();
+
+    /**
+     * Returns the number of packets received.
+     *
+     * @return the number of packets received
+     */
+    public long packetsReceived();
+
+    /**
+     * Returns the number of packets sent.
+     *
+     * @return the number of packets sent
+     */
+    public long packetsSent();
+
+    /**
+     * Returns the bytes received.
+     *
+     * @return the bytes received
+     */
+    public long bytesReceived();
+
+    /**
+     * Returns the bytes sent.
+     *
+     * @return the bytes sent
+     */
+    public long bytesSent();
+
+    /**
+     * Returns the number of packets dropped by RX.
+     *
+     * @return the number of packets dropped by RX
+     */
+    public long packetsRxDropped();
+
+    /**
+     * Returns the number of packets dropped by TX.
+     *
+     * @return the number of packets dropped by TX
+     */
+    public long packetsTxDropped();
+
+    /**
+     * Returns the number of transmit errors.
+     *
+     * @return the number of transmit errors
+     */
+    public long packetsRxErrors();
+
+    /**
+     * Returns the number of receive errors.
+     *
+     * @return the number of receive error
+     */
+    public long packetsTxErrors();
+
+    /**
+     * Returns the time port has been alive in seconds.
+     *
+     * @return the time port has been alive in seconds
+     */
+    public long durationSec();
+
+    /**
+     * Returns the time port has been alive in nano seconds.
+     *
+     * @return the time port has been alive in nano seconds
+     */
+    public long durationNano();
+
+}

@@ -69,6 +69,11 @@ public class DeviceServiceAdapter implements DeviceService {
     }
 
     @Override
+    public List<PortStatistics> getPortStatistics(DeviceId deviceId) {
+        return null;
+    }
+
+    @Override
     public Port getPort(DeviceId deviceId, PortNumber portNumber) {
         return null;
     }

@@ -50,10 +50,12 @@ import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.device.DeviceStore;
 import org.onosproject.net.device.DeviceStoreDelegate;
 import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.provider.AbstractProviderRegistry;
 import org.onosproject.net.provider.AbstractProviderService;
 import org.slf4j.Logger;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -174,6 +176,12 @@ public class DeviceManager
     }
 
     @Override
+    public List<PortStatistics> getPortStatistics(DeviceId deviceId) {
+        checkNotNull(deviceId, DEVICE_ID_NULL);
+        return store.getPortStatistics(deviceId);
+    }
+
+    @Override
     public Port getPort(DeviceId deviceId, PortNumber portNumber) {
         checkNotNull(deviceId, DEVICE_ID_NULL);
         checkNotNull(portNumber, PORT_NUMBER_NULL);
@@ -463,6 +471,19 @@ public class DeviceManager
             }
 
         }
+
+        @Override
+        public void updatePortStatistics(DeviceId deviceId, Collection<PortStatistics> portStatistics) {
+            checkNotNull(deviceId, DEVICE_ID_NULL);
+            checkNotNull(portStatistics,
+                    "Port statistics list cannot be null");
+            checkValidity();
+
+            DeviceEvent event = store.updatePortStatistics(this.provider().id(),
+                    deviceId, portStatistics);
+
+            post(event);
+        }
     }
 
     // Posts the specified event to the local event dispatcher.

@@ -53,6 +53,7 @@ import org.onosproject.net.device.DeviceEvent;
 import org.onosproject.net.device.DeviceStore;
 import org.onosproject.net.device.DeviceStoreDelegate;
 import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.store.AbstractStore;
 import org.onosproject.store.Timestamp;
@@ -121,6 +122,8 @@ 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();
 
     // to be updated under Device lock
     private final Map<DeviceId, Timestamp> offline = Maps.newHashMap();
@@ -800,6 +803,34 @@ public class GossipDeviceStore
     }
 
     @Override
+    public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
+                                            Collection<PortStatistics> portStats) {
+
+        ConcurrentMap<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId);
+        if (statsMap == null) {
+            statsMap = Maps.newConcurrentMap();
+            devicePortStats.put(deviceId, statsMap);
+        }
+
+        for (PortStatistics stat: portStats) {
+            PortNumber portNumber = PortNumber.portNumber(stat.port());
+            statsMap.put(portNumber, stat);
+        }
+
+        return 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();
+        }
+        return ImmutableList.copyOf(portStats.values());
+    }
+
+    @Override
     public Port getPort(DeviceId deviceId, PortNumber portNumber) {
         Map<PortNumber, Port> ports = devicePorts.get(deviceId);
         return ports == null ? null : ports.get(portNumber);

@@ -42,6 +42,7 @@ import org.onosproject.net.device.DeviceEvent;
 import org.onosproject.net.device.DeviceStore;
 import org.onosproject.net.device.DeviceStoreDelegate;
 import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.store.AbstractStore;
 import org.onlab.packet.ChassisId;
@@ -49,6 +50,7 @@ import org.onlab.util.NewConcurrentHashMap;
 import org.slf4j.Logger;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -94,6 +96,8 @@ public class SimpleDeviceStore
     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();
 
     // Available (=UP) devices
     private final Set<DeviceId> availableDevices = Sets.newConcurrentHashSet();
@@ -416,12 +420,39 @@ public class SimpleDeviceStore
     }
 
     @Override
+    public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
+                                            Collection<PortStatistics> portStats) {
+
+        ConcurrentMap<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId);
+        if (statsMap == null) {
+            statsMap = Maps.newConcurrentMap();
+            devicePortStats.put(deviceId, statsMap);
+        }
+
+        for (PortStatistics stat: portStats) {
+            PortNumber portNumber = PortNumber.portNumber(stat.port());
+            statsMap.put(portNumber, stat);
+        }
+
+        return new DeviceEvent(PORT_STATS_UPDATED,  devices.get(deviceId), null);
+    }
+
+    @Override
     public Port getPort(DeviceId deviceId, PortNumber portNumber) {
         Map<PortNumber, Port> ports = devicePorts.get(deviceId);
         return ports == null ? null : ports.get(portNumber);
     }
 
     @Override
+    public List<PortStatistics> getPortStatistics(DeviceId deviceId) {
+        Map<PortNumber, PortStatistics> portStats = devicePortStats.get(deviceId);
+        if (portStats == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableList.copyOf(portStats.values());
+    }
+
+    @Override
     public boolean isAvailable(DeviceId deviceId) {
         return availableDevices.contains(deviceId);
     }

@@ -47,6 +47,8 @@ import org.projectfloodlight.openflow.protocol.OFGroupStatsReply;
 import org.projectfloodlight.openflow.protocol.OFMessage;
 import org.projectfloodlight.openflow.protocol.OFPacketIn;
 import org.projectfloodlight.openflow.protocol.OFPortDesc;
+import org.projectfloodlight.openflow.protocol.OFPortStatsEntry;
+import org.projectfloodlight.openflow.protocol.OFPortStatsReply;
 import org.projectfloodlight.openflow.protocol.OFPortStatus;
 import org.projectfloodlight.openflow.protocol.OFStatsReply;
 import org.projectfloodlight.openflow.protocol.OFStatsReplyFlags;
@@ -104,6 +106,9 @@ public class OpenFlowControllerImpl implements OpenFlowController {
     protected Multimap<Dpid, OFGroupDescStatsEntry> fullGroupDescStats =
             ArrayListMultimap.create();
 
+    protected Multimap<Dpid, OFPortStatsEntry> fullPortStats =
+            ArrayListMultimap.create();
+
     private final Controller ctrl = new Controller();
 
     @Activate
@@ -216,6 +221,7 @@ public class OpenFlowControllerImpl implements OpenFlowController {
         Collection<OFFlowStatsEntry> flowStats;
         Collection<OFGroupStatsEntry> groupStats;
         Collection<OFGroupDescStatsEntry> groupDescStats;
+        Collection<OFPortStatsEntry> portStats;
 
         switch (msg.getType()) {
         case PORT_STATUS:
@@ -280,6 +286,9 @@ public class OpenFlowControllerImpl implements OpenFlowController {
                         executorMsgs.submit(new OFMessageHandler(dpid, rep.build()));
                     }
                     break;
+                case PORT:
+                    executorMsgs.submit(new OFMessageHandler(dpid, reply));
+                    break;
                 default:
                     log.warn("Unsupported stats type : {}", reply.getStatsType());
             }
@@ -343,6 +352,15 @@ public class OpenFlowControllerImpl implements OpenFlowController {
         return null;
     }
 
+    private synchronized Collection<OFPortStatsEntry> publishPortStats(Dpid dpid,
+                                                                 OFPortStatsReply reply) {
+        fullPortStats.putAll(dpid, reply.getEntries());
+        if (!reply.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
+            return fullPortStats.removeAll(dpid);
+        }
+        return null;
+    }
+
     @Override
     public void setRole(Dpid dpid, RoleState role) {
         final OpenFlowSwitch sw = getSwitch(dpid);

@@ -15,6 +15,8 @@
  */
 package org.onosproject.provider.of.device.impl;
 
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -29,31 +31,43 @@ import org.onosproject.net.PortNumber;
 import org.onosproject.net.SparseAnnotations;
 import org.onosproject.net.device.DefaultDeviceDescription;
 import org.onosproject.net.device.DefaultPortDescription;
+import org.onosproject.net.device.DefaultPortStatistics;
 import org.onosproject.net.device.DeviceDescription;
 import org.onosproject.net.device.DeviceProvider;
 import org.onosproject.net.device.DeviceProviderRegistry;
 import org.onosproject.net.device.DeviceProviderService;
 import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.provider.AbstractProvider;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.openflow.controller.Dpid;
 import org.onosproject.openflow.controller.OpenFlowController;
+import org.onosproject.openflow.controller.OpenFlowEventListener;
 import org.onosproject.openflow.controller.OpenFlowSwitch;
 import org.onosproject.openflow.controller.OpenFlowSwitchListener;
 import org.onosproject.openflow.controller.RoleState;
 import org.onlab.packet.ChassisId;
 import org.projectfloodlight.openflow.protocol.OFFactory;
+import org.projectfloodlight.openflow.protocol.OFMessage;
 import org.projectfloodlight.openflow.protocol.OFPortConfig;
 import org.projectfloodlight.openflow.protocol.OFPortDesc;
 import org.projectfloodlight.openflow.protocol.OFPortFeatures;
 import org.projectfloodlight.openflow.protocol.OFPortReason;
 import org.projectfloodlight.openflow.protocol.OFPortState;
+import org.projectfloodlight.openflow.protocol.OFPortStatsEntry;
+import org.projectfloodlight.openflow.protocol.OFPortStatsReply;
 import org.projectfloodlight.openflow.protocol.OFPortStatus;
+import org.projectfloodlight.openflow.protocol.OFStatsReply;
+import org.projectfloodlight.openflow.protocol.OFStatsType;
 import org.projectfloodlight.openflow.protocol.OFVersion;
 import org.projectfloodlight.openflow.types.PortSpeed;
 import org.slf4j.Logger;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 
 import com.google.common.base.Strings;
@@ -83,7 +97,12 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
 
     private DeviceProviderService providerService;
 
-    private final OpenFlowSwitchListener listener = new InternalDeviceProvider();
+    private final InternalDeviceProvider listener = new InternalDeviceProvider();
+
+    // TODO: We need to make the poll interval configurable.
+    static final int POLL_INTERVAL = 10;
+
+    private HashMap<Dpid, PortStatsCollector> collectors = Maps.newHashMap();
 
     /**
      * Creates an OpenFlow device provider.
@@ -96,6 +115,7 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
     public void activate() {
         providerService = providerRegistry.register(this);
         controller.addListener(listener);
+        controller.addEventListener(listener);
         for (OpenFlowSwitch sw : controller.getSwitches()) {
             try {
                 listener.switchAdded(new Dpid(sw.getId()));
@@ -105,6 +125,9 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
                 // disconnect to trigger switch-add later
                 sw.disconnectSwitch();
             }
+            PortStatsCollector psc = new PortStatsCollector(sw, POLL_INTERVAL);
+            psc.start();
+            collectors.put(new Dpid(sw.getId()), psc);
         }
         LOG.info("Started");
     }
@@ -174,7 +197,45 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
         LOG.info("Accepting mastership role change for device {}", deviceId);
     }
 
-    private class InternalDeviceProvider implements OpenFlowSwitchListener {
+    private void pushPortMetrics(Dpid dpid, OFPortStatsReply msg) {
+        DeviceId deviceId = DeviceId.deviceId(dpid.uri(dpid));
+
+        Collection<PortStatistics> stats = buildPortStatistics(deviceId, msg);
+
+        providerService.updatePortStatistics(deviceId, stats);
+    }
+
+    private Collection<PortStatistics> buildPortStatistics(DeviceId deviceId, OFPortStatsReply msg) {
+
+        HashSet<PortStatistics> stats = Sets.newHashSet();
+
+        for (OFPortStatsEntry entry: msg.getEntries()) {
+            if (entry.getPortNo().getPortNumber() < 0) {
+                continue;
+            }
+            DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
+            DefaultPortStatistics stat = builder.setDeviceId(deviceId)
+                    .setPort(entry.getPortNo().getPortNumber())
+                    .setPacketsReceived(entry.getRxPackets().getValue())
+                    .setPacketsSent(entry.getTxPackets().getValue())
+                    .setBytesReceived(entry.getRxBytes().getValue())
+                    .setBytesSent(entry.getTxBytes().getValue())
+                    .setPacketsRxDropped(entry.getRxDropped().getValue())
+                    .setPacketsTxDropped(entry.getTxDropped().getValue())
+                    .setPacketsRxErrors(entry.getRxErrors().getValue())
+                    .setPacketsTxErrors(entry.getTxErrors().getValue())
+                    .setDurationSec(entry.getDurationSec())
+                    .setDurationNano(entry.getDurationNsec())
+                    .build();
+
+            stats.add(stat);
+        }
+
+        return Collections.unmodifiableSet(stats);
+
+    }
+
+    private class InternalDeviceProvider implements OpenFlowSwitchListener, OpenFlowEventListener {
         @Override
         public void switchAdded(Dpid dpid) {
             if (providerService == null) {
@@ -201,6 +262,11 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
                                                  cId, annotations);
             providerService.deviceConnected(did, description);
             providerService.updatePorts(did, buildPortDescriptions(sw.getPorts()));
+
+            PortStatsCollector psc = new PortStatsCollector(
+                        controller.getSwitch(dpid), POLL_INTERVAL);
+            psc.start();
+            collectors.put(dpid, psc);
         }
 
         @Override
@@ -209,8 +275,12 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
                 return;
             }
             providerService.deviceDisconnected(deviceId(uri(dpid)));
-        }
 
+            PortStatsCollector collector = collectors.remove(dpid);
+            if (collector != null) {
+                collector.stop();
+            }
+        }
 
         @Override
         public void switchChanged(Dpid dpid) {
@@ -328,6 +398,19 @@ public class OpenFlowDeviceProvider extends AbstractProvider implements DevicePr
             }
             return portSpeed.getSpeedBps() / MBPS;
         }
+
+        @Override
+        public void handleMessage(Dpid dpid, OFMessage msg) {
+            switch (msg.getType()) {
+                case STATS_REPLY:
+                    if (((OFStatsReply) msg).getStatsType() == OFStatsType.PORT) {
+                        pushPortMetrics(dpid, (OFPortStatsReply) msg);
+                    }
+                    break;
+                default:
+                    break;
+            }
+        }
     }
 
 }

+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.provider.of.device.impl;
+
+import org.jboss.netty.util.HashedWheelTimer;
+import org.jboss.netty.util.Timeout;
+import org.jboss.netty.util.TimerTask;
+import org.onlab.util.Timer;
+import org.onosproject.openflow.controller.OpenFlowSwitch;
+import org.onosproject.openflow.controller.RoleState;
+import org.projectfloodlight.openflow.protocol.OFPortStatsRequest;
+import org.projectfloodlight.openflow.types.OFPort;
+import org.slf4j.Logger;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/*
+ * Sends Group Stats Request and collect the group statistics with a time interval.
+ */
+public class PortStatsCollector implements TimerTask {
+
+    // TODO: Refactoring is required using ScheduledExecutorService
+
+    private final HashedWheelTimer timer = Timer.getTimer();
+    private final OpenFlowSwitch sw;
+    private final Logger log = getLogger(getClass());
+    private final int refreshInterval;
+    private final AtomicLong xidAtomic = new AtomicLong(1);
+
+    private Timeout timeout;
+
+    private boolean stopTimer = false;
+
+    /**
+     * Creates a GroupStatsCollector object.
+     *
+     * @param sw Open Flow switch
+     * @param interval time interval for collecting group statistic
+     */
+    public PortStatsCollector(OpenFlowSwitch sw, int interval) {
+        this.sw = sw;
+        this.refreshInterval = interval;
+    }
+
+    @Override
+    public void run(Timeout timeout) throws Exception {
+        log.trace("Collecting stats for {}", sw.getStringId());
+
+        sendPortStatistic();
+
+        if (!this.stopTimer) {
+            log.trace("Scheduling stats collection in {} seconds for {}",
+                    this.refreshInterval, this.sw.getStringId());
+            timeout.getTimer().newTimeout(this, refreshInterval,
+                    TimeUnit.SECONDS);
+        }
+    }
+
+    private void sendPortStatistic() {
+        if (log.isTraceEnabled()) {
+            log.trace("sendGroupStatistics {}:{}", sw.getStringId(), sw.getRole());
+        }
+        if (sw.getRole() != RoleState.MASTER) {
+            return;
+        }
+        Long statsXid = xidAtomic.getAndIncrement();
+        OFPortStatsRequest statsRequest = sw.factory().buildPortStatsRequest()
+                .setPortNo(OFPort.ANY)
+                .setXid(statsXid)
+                .build();
+        sw.sendMsg(statsRequest);
+    }
+
+    /**
+     * Starts the collector.
+     */
+    public void start() {
+        log.info("Starting Port Stats collection thread for {}", sw.getStringId());
+        timeout = timer.newTimeout(this, 1, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Stops the collector.
+     */
+    public void stop() {
+        log.info("Stopping Port Stats collection thread for {}", sw.getStringId());
+        this.stopTimer = true;
+        timeout.cancel();
+    }
+}

@@ -23,6 +23,7 @@ import static org.onosproject.net.Device.Type.*;
 import static org.onosproject.net.MastershipRole.*;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -41,6 +42,7 @@ import org.onosproject.net.device.DeviceProvider;
 import org.onosproject.net.device.DeviceProviderRegistry;
 import org.onosproject.net.device.DeviceProviderService;
 import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.openflow.controller.Dpid;
 import org.onosproject.openflow.controller.OpenFlowController;
@@ -217,6 +219,11 @@ public class OpenFlowDeviceProviderTest {
                 roles.put(requested, Dpid.dpid(deviceId.uri()));
             }
 
+            @Override
+            public void updatePortStatistics(DeviceId deviceId, Collection<PortStatistics> portStatistics) {
+
+            }
+
         }
     }