홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Dusan Pajin
Committed by Gerrit Code Review
11ff4a8f 1 parent 25a2bf4f

Delta Port Statistics added to DeviceStore, Device Service and CLI 

Changed misspelled word in description of the portstats command switch

Change-Id: I131940c83c5cb12080532a4804ec424ca66afa64
Inline Side-by-side
Showing 7 changed files with 318 additions and 28 deletions
...@@ -16,8 +16,14 @@ package org.onosproject.cli.net; ...@@ -16,8 +16,14 @@ package org.onosproject.cli.net;
16 * limitations under the License. 16 * limitations under the License.
17 */ 17 */
18 18
19 +import static org.onosproject.net.DeviceId.deviceId;
20 +
21 +import java.util.concurrent.TimeUnit;
22 +
23 +import org.apache.karaf.shell.commands.Argument;
19 import org.apache.karaf.shell.commands.Command; 24 import org.apache.karaf.shell.commands.Command;
20 -import org.onosproject.cli.AbstractShellCommand; 25 +import org.apache.karaf.shell.commands.Option;
26 +import org.onosproject.net.Device;
21 import org.onosproject.net.DeviceId; 27 import org.onosproject.net.DeviceId;
22 import org.onosproject.net.device.DeviceService; 28 import org.onosproject.net.device.DeviceService;
23 import org.onosproject.net.device.PortStatistics; 29 import org.onosproject.net.device.PortStatistics;
...@@ -27,7 +33,20 @@ import org.onosproject.net.device.PortStatistics; ...@@ -27,7 +33,20 @@ import org.onosproject.net.device.PortStatistics;
27 */ 33 */
28 @Command(scope = "onos", name = "portstats", 34 @Command(scope = "onos", name = "portstats",
29 description = "Lists statistics of all ports in the system") 35 description = "Lists statistics of all ports in the system")
30 -public class DevicePortStatsCommand extends AbstractShellCommand { 36 +public class DevicePortStatsCommand extends DevicesListCommand {
37 +
38 + @Option(name = "-d", aliases = "--delta", description = "Show Delta Port Statistics,"
39 + + "only for the last polling interval",
40 + required = false, multiValued = false)
41 + private boolean delta = false;
42 +
43 + @Option(name = "-t", aliases = "--table", description = "Show human readable table format for statistics",
44 + required = false, multiValued = false)
45 + private boolean table = false;
46 +
47 + @Argument(index = 0, name = "uri", description = "Device ID",
48 + required = false, multiValued = false)
49 + String uri = null;
31 50
32 private static final String FORMAT = 51 private static final String FORMAT =
33 " port=%s, pktRx=%s, pktTx=%s, bytesRx=%s, bytesTx=%s, pktRxDrp=%s, pktTxDrp=%s, Dur=%s"; 52 " port=%s, pktRx=%s, pktTx=%s, bytesRx=%s, bytesTx=%s, pktRxDrp=%s, pktTxDrp=%s, Dur=%s";
...@@ -36,16 +55,138 @@ public class DevicePortStatsCommand extends AbstractShellCommand { ...@@ -36,16 +55,138 @@ public class DevicePortStatsCommand extends AbstractShellCommand {
36 protected void execute() { 55 protected void execute() {
37 DeviceService deviceService = get(DeviceService.class); 56 DeviceService deviceService = get(DeviceService.class);
38 57
39 - deviceService.getDevices().forEach(d -> 58 + if (uri == null) {
40 - printPortStats(d.id(), deviceService.getPortStatistics(d.id())) 59 + for (Device d : getSortedDevices(deviceService)) {
41 - ); 60 + if (delta) {
61 + if (table) {
62 + printPortStatsDeltaTable(d.id(), deviceService.getPortDeltaStatistics(d.id()));
63 + } else {
64 + printPortStatsDelta(d.id(), deviceService.getPortDeltaStatistics(d.id()));
65 + }
66 + } else {
67 + printPortStats(d.id(), deviceService.getPortStatistics(d.id()));
68 + }
69 + }
70 + } else {
71 + Device d = deviceService.getDevice(deviceId(uri));
72 + if (d == null) {
73 + error("No such device %s", uri);
74 + } else if (delta) {
75 + if (table) {
76 + printPortStatsDeltaTable(d.id(), deviceService.getPortDeltaStatistics(d.id()));
77 + } else {
78 + printPortStatsDelta(d.id(), deviceService.getPortDeltaStatistics(d.id()));
79 + }
80 + } else {
81 + printPortStats(d.id(), deviceService.getPortStatistics(d.id()));
82 + }
83 + }
42 } 84 }
43 85
86 + /**
87 + * Prints Port Statistics.
88 + *
89 + * @param deviceId
90 + * @param portStats
91 + */
44 private void printPortStats(DeviceId deviceId, Iterable<PortStatistics> portStats) { 92 private void printPortStats(DeviceId deviceId, Iterable<PortStatistics> portStats) {
93 +
45 print("deviceId=%s", deviceId); 94 print("deviceId=%s", deviceId);
46 for (PortStatistics stat : portStats) { 95 for (PortStatistics stat : portStats) {
47 print(FORMAT, stat.port(), stat.packetsReceived(), stat.packetsSent(), stat.bytesReceived(), 96 print(FORMAT, stat.port(), stat.packetsReceived(), stat.packetsSent(), stat.bytesReceived(),
48 stat.bytesSent(), stat.packetsRxDropped(), stat.packetsTxDropped(), stat.durationSec()); 97 stat.bytesSent(), stat.packetsRxDropped(), stat.packetsTxDropped(), stat.durationSec());
49 } 98 }
50 } 99 }
100 + /**
101 + * Prints Port delta statistics.
102 + *
103 + * @param deviceId
104 + * @param portStats
105 + */
106 + private void printPortStatsDelta(DeviceId deviceId, Iterable<PortStatistics> portStats) {
107 + final String formatDelta = " port=%s, pktRx=%s, pktTx=%s, bytesRx=%s, bytesTx=%s,"
108 + + " rateRx=%s, rateTx=%s, pktRxDrp=%s, pktTxDrp=%s, interval=%s";
109 + print("deviceId=%s", deviceId);
110 + for (PortStatistics stat : portStats) {
111 + float duration = ((float) stat.durationSec()) +
112 + (((float) stat.durationNano()) / TimeUnit.SECONDS.toNanos(1));
113 + float rateRx = stat.bytesReceived() * 8 / duration;
114 + float rateTx = stat.bytesSent() * 8 / duration;
115 + print(formatDelta, stat.port(),
116 + stat.packetsReceived(),
117 + stat.packetsSent(),
118 + stat.bytesReceived(),
119 + stat.bytesSent(),
120 + String.format("%.1f", rateRx),
121 + String.format("%.1f", rateTx),
122 + stat.packetsRxDropped(),
123 + stat.packetsTxDropped(),
124 + String.format("%.3f", duration));
125 + }
126 + }
127 +
128 + /**
129 + * Prints human readable table with delta Port Statistics for specific device.
130 + *
131 + * @param deviceId
132 + * @param portStats
133 + */
134 + private void printPortStatsDeltaTable(DeviceId deviceId, Iterable<PortStatistics> portStats) {
135 + final String formatDeltaTable = "|%5s | %7s | %7s | %7s | %7s | %7s | %7s | %7s | %7s |%9s |";
136 + print("+---------------------------------------------------------------------------------------------------+");
137 + print("| DeviceId = %s |", deviceId);
138 + print("|---------------------------------------------------------------------------------------------------|");
139 + print("| | Receive | Transmit | Time [s] |");
140 + print("| Port | Packets | Bytes | Rate bps | Drop | Packets | Bytes | Rate bps | Drop | Interval |");
141 + print("|---------------------------------------------------------------------------------------------------|");
142 +
143 + for (PortStatistics stat : portStats) {
144 + float duration = ((float) stat.durationSec()) +
145 + (((float) stat.durationNano()) / TimeUnit.SECONDS.toNanos(1));
146 + float rateRx = stat.bytesReceived() * 8 / duration;
147 + float rateTx = stat.bytesSent() * 8 / duration;
148 + print(formatDeltaTable, stat.port(),
149 + humanReadable(stat.packetsReceived()),
150 + humanReadable(stat.bytesReceived()),
151 + humanReadableBps(rateRx),
152 + humanReadable(stat.packetsRxDropped()),
153 + humanReadable(stat.packetsSent()),
154 + humanReadable(stat.bytesSent()),
155 + humanReadableBps(rateTx),
156 + humanReadable(stat.packetsTxDropped()),
157 + String.format("%.3f", duration));
158 + }
159 + print("+---------------------------------------------------------------------------------------------------+");
160 + }
161 +
162 + /**
163 + * Converts bytes to human readable string with Kilo, Mega, Giga, etc.
164 + *
165 + * @param bytes
166 + * @return
167 + */
168 + public static String humanReadable(long bytes) {
169 + int unit = 1000;
170 + if (bytes < unit) {
171 + return String.format("%s ", bytes);
172 + }
173 + int exp = (int) (Math.log(bytes) / Math.log(unit));
174 + Character pre = ("KMGTPE").charAt(exp - 1);
175 + return String.format("%.2f%s", bytes / Math.pow(unit, exp), pre);
176 + }
177 + /**
178 + * Converts bps to human readable format.
179 + *
180 + * @param bps
181 + * @return
182 + */
183 + public static String humanReadableBps(float bps) {
184 + int unit = 1000;
185 + if (bps < unit) {
186 + return String.format("%.0f ", (float) bps);
187 + }
188 + int exp = (int) (Math.log(bps) / Math.log(unit));
189 + Character pre = ("KMGTPE").charAt(exp - 1);
190 + return String.format("%.2f%s", bps / Math.pow(unit, exp), pre);
191 + }
51 } 192 }
......
...@@ -105,6 +105,14 @@ public interface DeviceService ...@@ -105,6 +105,14 @@ public interface DeviceService
105 List<PortStatistics> getPortStatistics(DeviceId deviceId); 105 List<PortStatistics> getPortStatistics(DeviceId deviceId);
106 106
107 /** 107 /**
108 + * Returns the list of port delta statistics associated with the device.
109 + *
110 + * @param deviceId device identitifer
111 + * @return list of port statistics
112 + */
113 + List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId);
114 +
115 + /**
108 * Returns the port with the specified number and hosted by the given device. 116 * Returns the port with the specified number and hosted by the given device.
109 * 117 *
110 * @param deviceId device identifier 118 * @param deviceId device identifier
......
...@@ -135,6 +135,14 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> { ...@@ -135,6 +135,14 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> {
135 List<PortStatistics> getPortStatistics(DeviceId deviceId); 135 List<PortStatistics> getPortStatistics(DeviceId deviceId);
136 136
137 /** 137 /**
138 + * Returns the list of delta port statistics of the specified device.
139 + *
140 + * @param deviceId
141 + * @return list of delta port statistics of all ports of the device
142 + */
143 + List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId);
144 +
145 + /**
138 * Returns the specified device port. 146 * Returns the specified device port.
139 * 147 *
140 * @param deviceId device identifier 148 * @param deviceId device identifier
...@@ -159,5 +167,4 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> { ...@@ -159,5 +167,4 @@ public interface DeviceStore extends Store<DeviceEvent, DeviceStoreDelegate> {
159 */ 167 */
160 DeviceEvent removeDevice(DeviceId deviceId); 168 DeviceEvent removeDevice(DeviceId deviceId);
161 169
162 -
163 } 170 }
......
...@@ -75,6 +75,11 @@ public class DeviceServiceAdapter implements DeviceService { ...@@ -75,6 +75,11 @@ public class DeviceServiceAdapter implements DeviceService {
75 } 75 }
76 76
77 @Override 77 @Override
78 + public List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId) {
79 + return null;
80 + }
81 +
82 + @Override
78 public Port getPort(DeviceId deviceId, PortNumber portNumber) { 83 public Port getPort(DeviceId deviceId, PortNumber portNumber) {
79 return null; 84 return null;
80 } 85 }
......
...@@ -37,6 +37,7 @@ import org.onosproject.net.PortNumber; ...@@ -37,6 +37,7 @@ import org.onosproject.net.PortNumber;
37 import org.onosproject.net.SparseAnnotations; 37 import org.onosproject.net.SparseAnnotations;
38 import org.onosproject.net.device.DefaultDeviceDescription; 38 import org.onosproject.net.device.DefaultDeviceDescription;
39 import org.onosproject.net.device.DefaultPortDescription; 39 import org.onosproject.net.device.DefaultPortDescription;
40 +import org.onosproject.net.device.DefaultPortStatistics;
40 import org.onosproject.net.device.DeviceDescription; 41 import org.onosproject.net.device.DeviceDescription;
41 import org.onosproject.net.device.DeviceEvent; 42 import org.onosproject.net.device.DeviceEvent;
42 import org.onosproject.net.device.DeviceStore; 43 import org.onosproject.net.device.DeviceStore;
...@@ -62,6 +63,7 @@ import java.util.Objects; ...@@ -62,6 +63,7 @@ import java.util.Objects;
62 import java.util.Set; 63 import java.util.Set;
63 import java.util.concurrent.ConcurrentHashMap; 64 import java.util.concurrent.ConcurrentHashMap;
64 import java.util.concurrent.ConcurrentMap; 65 import java.util.concurrent.ConcurrentMap;
66 +import java.util.concurrent.TimeUnit;
65 import java.util.concurrent.atomic.AtomicReference; 67 import java.util.concurrent.atomic.AtomicReference;
66 68
67 import static com.google.common.base.Preconditions.checkArgument; 69 import static com.google.common.base.Preconditions.checkArgument;
...@@ -98,6 +100,8 @@ public class SimpleDeviceStore ...@@ -98,6 +100,8 @@ public class SimpleDeviceStore
98 devicePorts = Maps.newConcurrentMap(); 100 devicePorts = Maps.newConcurrentMap();
99 private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, PortStatistics>> 101 private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, PortStatistics>>
100 devicePortStats = Maps.newConcurrentMap(); 102 devicePortStats = Maps.newConcurrentMap();
103 + private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, PortStatistics>>
104 + devicePortDeltaStats = Maps.newConcurrentMap();
101 105
102 // Available (=UP) devices 106 // Available (=UP) devices
103 private final Set<DeviceId> availableDevices = Sets.newConcurrentHashSet(); 107 private final Set<DeviceId> availableDevices = Sets.newConcurrentHashSet();
...@@ -421,22 +425,62 @@ public class SimpleDeviceStore ...@@ -421,22 +425,62 @@ public class SimpleDeviceStore
421 425
422 @Override 426 @Override
423 public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId, 427 public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
424 - Collection<PortStatistics> portStats) { 428 + Collection<PortStatistics> newStatsCollection) {
425 429
426 - ConcurrentMap<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId); 430 + ConcurrentMap<PortNumber, PortStatistics> prvStatsMap = devicePortStats.get(deviceId);
427 - if (statsMap == null) { 431 + ConcurrentMap<PortNumber, PortStatistics> newStatsMap = Maps.newConcurrentMap();
428 - statsMap = Maps.newConcurrentMap(); 432 + ConcurrentMap<PortNumber, PortStatistics> deltaStatsMap = Maps.newConcurrentMap();
429 - devicePortStats.put(deviceId, statsMap);
430 - }
431 433
432 - for (PortStatistics stat: portStats) { 434 + if (prvStatsMap != null) {
433 - PortNumber portNumber = PortNumber.portNumber(stat.port()); 435 + for (PortStatistics newStats : newStatsCollection) {
434 - statsMap.put(portNumber, stat); 436 + PortNumber port = PortNumber.portNumber(newStats.port());
437 + PortStatistics prvStats = prvStatsMap.get(port);
438 + DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
439 + PortStatistics deltaStats = builder.build();
440 + if (prvStats != null) {
441 + deltaStats = calcDeltaStats(deviceId, prvStats, newStats);
435 } 442 }
436 - 443 + deltaStatsMap.put(port, deltaStats);
444 + newStatsMap.put(port, newStats);
445 + }
446 + } else {
447 + for (PortStatistics newStats : newStatsCollection) {
448 + PortNumber port = PortNumber.portNumber(newStats.port());
449 + newStatsMap.put(port, newStats);
450 + }
451 + }
452 + devicePortDeltaStats.put(deviceId, deltaStatsMap);
453 + devicePortStats.put(deviceId, newStatsMap);
437 return new DeviceEvent(PORT_STATS_UPDATED, devices.get(deviceId), null); 454 return new DeviceEvent(PORT_STATS_UPDATED, devices.get(deviceId), null);
438 } 455 }
439 456
457 + public PortStatistics calcDeltaStats(DeviceId deviceId, PortStatistics prvStats, PortStatistics newStats) {
458 + // calculate time difference
459 + long deltaStatsSec, deltaStatsNano;
460 + if (newStats.durationNano() < prvStats.durationNano()) {
461 + deltaStatsNano = newStats.durationNano() - prvStats.durationNano() + TimeUnit.SECONDS.toNanos(1);
462 + deltaStatsSec = newStats.durationSec() - prvStats.durationSec() - 1L;
463 + } else {
464 + deltaStatsNano = newStats.durationNano() - prvStats.durationNano();
465 + deltaStatsSec = newStats.durationSec() - prvStats.durationSec();
466 + }
467 + DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
468 + DefaultPortStatistics deltaStats = builder.setDeviceId(deviceId)
469 + .setPort(newStats.port())
470 + .setPacketsReceived(newStats.packetsReceived() - prvStats.packetsReceived())
471 + .setPacketsSent(newStats.packetsSent() - prvStats.packetsSent())
472 + .setBytesReceived(newStats.bytesReceived() - prvStats.bytesReceived())
473 + .setBytesSent(newStats.bytesSent() - prvStats.bytesSent())
474 + .setPacketsRxDropped(newStats.packetsRxDropped() - prvStats.packetsRxDropped())
475 + .setPacketsTxDropped(newStats.packetsTxDropped() - prvStats.packetsTxDropped())
476 + .setPacketsRxErrors(newStats.packetsRxErrors() - prvStats.packetsRxErrors())
477 + .setPacketsTxErrors(newStats.packetsTxErrors() - prvStats.packetsTxErrors())
478 + .setDurationSec(deltaStatsSec)
479 + .setDurationNano(deltaStatsNano)
480 + .build();
481 + return deltaStats;
482 + }
483 +
440 @Override 484 @Override
441 public Port getPort(DeviceId deviceId, PortNumber portNumber) { 485 public Port getPort(DeviceId deviceId, PortNumber portNumber) {
442 Map<PortNumber, Port> ports = devicePorts.get(deviceId); 486 Map<PortNumber, Port> ports = devicePorts.get(deviceId);
...@@ -453,6 +497,15 @@ public class SimpleDeviceStore ...@@ -453,6 +497,15 @@ public class SimpleDeviceStore
453 } 497 }
454 498
455 @Override 499 @Override
500 + public List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId) {
501 + Map<PortNumber, PortStatistics> portStats = devicePortDeltaStats.get(deviceId);
502 + if (portStats == null) {
503 + return Collections.emptyList();
504 + }
505 + return ImmutableList.copyOf(portStats.values());
506 + }
507 +
508 + @Override
456 public boolean isAvailable(DeviceId deviceId) { 509 public boolean isAvailable(DeviceId deviceId) {
457 return availableDevices.contains(deviceId); 510 return availableDevices.contains(deviceId);
458 } 511 }
......
...@@ -196,6 +196,13 @@ public class DeviceManager ...@@ -196,6 +196,13 @@ public class DeviceManager
196 } 196 }
197 197
198 @Override 198 @Override
199 + public List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId) {
200 + checkPermission(Permission.DEVICE_READ);
201 + checkNotNull(deviceId, DEVICE_ID_NULL);
202 + return store.getPortDeltaStatistics(deviceId);
203 + }
204 +
205 + @Override
199 public Port getPort(DeviceId deviceId, PortNumber portNumber) { 206 public Port getPort(DeviceId deviceId, PortNumber portNumber) {
200 checkPermission(Permission.DEVICE_READ); 207 checkPermission(Permission.DEVICE_READ);
201 checkNotNull(deviceId, DEVICE_ID_NULL); 208 checkNotNull(deviceId, DEVICE_ID_NULL);
......
...@@ -20,6 +20,7 @@ import com.google.common.collect.FluentIterable; ...@@ -20,6 +20,7 @@ import com.google.common.collect.FluentIterable;
20 import com.google.common.collect.ImmutableList; 20 import com.google.common.collect.ImmutableList;
21 import com.google.common.collect.Maps; 21 import com.google.common.collect.Maps;
22 import com.google.common.collect.Sets; 22 import com.google.common.collect.Sets;
23 +
23 import org.apache.commons.lang3.RandomUtils; 24 import org.apache.commons.lang3.RandomUtils;
24 import org.apache.felix.scr.annotations.Activate; 25 import org.apache.felix.scr.annotations.Activate;
25 import org.apache.felix.scr.annotations.Component; 26 import org.apache.felix.scr.annotations.Component;
...@@ -108,7 +109,6 @@ import static org.onlab.util.Tools.minPriority; ...@@ -108,7 +109,6 @@ import static org.onlab.util.Tools.minPriority;
108 import static org.onosproject.cluster.ControllerNodeToNodeId.toNodeId; 109 import static org.onosproject.cluster.ControllerNodeToNodeId.toNodeId;
109 import static org.onosproject.net.DefaultAnnotations.merge; 110 import static org.onosproject.net.DefaultAnnotations.merge;
110 import static org.onosproject.net.device.DeviceEvent.Type.*; 111 import static org.onosproject.net.device.DeviceEvent.Type.*;
111 -import static org.onosproject.net.device.DeviceEvent.Type.DEVICE_REMOVED;
112 import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.*; 112 import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.*;
113 import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT; 113 import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
114 import static org.slf4j.LoggerFactory.getLogger; 114 import static org.slf4j.LoggerFactory.getLogger;
...@@ -139,6 +139,7 @@ public class GossipDeviceStore ...@@ -139,6 +139,7 @@ public class GossipDeviceStore
139 private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = Maps.newConcurrentMap(); 139 private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = Maps.newConcurrentMap();
140 140
141 private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortStats; 141 private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortStats;
142 + private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortDeltaStats;
142 private final EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>> 143 private final EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>>
143 portStatsListener = new InternalPortStatsListener(); 144 portStatsListener = new InternalPortStatsListener();
144 145
...@@ -246,6 +247,14 @@ public class GossipDeviceStore ...@@ -246,6 +247,14 @@ public class GossipDeviceStore
246 .withTimestampProvider((k, v) -> new WallClockTimestamp()) 247 .withTimestampProvider((k, v) -> new WallClockTimestamp())
247 .withTombstonesDisabled() 248 .withTombstonesDisabled()
248 .build(); 249 .build();
250 + devicePortDeltaStats = storageService.<DeviceId, Map<PortNumber, PortStatistics>>
251 + eventuallyConsistentMapBuilder()
252 + .withName("port-stats-delta")
253 + .withSerializer(deviceDataSerializer)
254 + .withAntiEntropyPeriod(5, TimeUnit.SECONDS)
255 + .withTimestampProvider((k, v) -> new WallClockTimestamp())
256 + .withTombstonesDisabled()
257 + .build();
249 devicePortStats.addListener(portStatsListener); 258 devicePortStats.addListener(portStatsListener);
250 log.info("Started"); 259 log.info("Started");
251 } 260 }
...@@ -253,6 +262,7 @@ public class GossipDeviceStore ...@@ -253,6 +262,7 @@ public class GossipDeviceStore
253 @Deactivate 262 @Deactivate
254 public void deactivate() { 263 public void deactivate() {
255 devicePortStats.destroy(); 264 devicePortStats.destroy();
265 + devicePortDeltaStats.destroy();
256 executor.shutdownNow(); 266 executor.shutdownNow();
257 267
258 backgroundExecutor.shutdownNow(); 268 backgroundExecutor.shutdownNow();
...@@ -824,18 +834,68 @@ public class GossipDeviceStore ...@@ -824,18 +834,68 @@ public class GossipDeviceStore
824 834
825 @Override 835 @Override
826 public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId, 836 public DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
827 - Collection<PortStatistics> portStats) { 837 + Collection<PortStatistics> newStatsCollection) {
828 - Map<PortNumber, PortStatistics> statsMap = devicePortStats.get(deviceId);
829 - if (statsMap == null) {
830 - statsMap = Maps.newHashMap();
831 - }
832 838
833 - for (PortStatistics stat : portStats) { 839 + Map<PortNumber, PortStatistics> prvStatsMap = devicePortStats.get(deviceId);
834 - PortNumber portNumber = PortNumber.portNumber(stat.port()); 840 + Map<PortNumber, PortStatistics> newStatsMap = Maps.newHashMap();
835 - statsMap.put(portNumber, stat); 841 + Map<PortNumber, PortStatistics> deltaStatsMap = Maps.newHashMap();
842 +
843 + if (prvStatsMap != null) {
844 + for (PortStatistics newStats : newStatsCollection) {
845 + PortNumber port = PortNumber.portNumber(newStats.port());
846 + PortStatistics prvStats = prvStatsMap.get(port);
847 + DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
848 + PortStatistics deltaStats = builder.build();
849 + if (prvStats != null) {
850 + deltaStats = calcDeltaStats(deviceId, prvStats, newStats);
851 + }
852 + deltaStatsMap.put(port, deltaStats);
853 + newStatsMap.put(port, newStats);
854 + }
855 + } else {
856 + for (PortStatistics newStats : newStatsCollection) {
857 + PortNumber port = PortNumber.portNumber(newStats.port());
858 + newStatsMap.put(port, newStats);
859 + }
860 + }
861 + devicePortDeltaStats.put(deviceId, deltaStatsMap);
862 + devicePortStats.put(deviceId, newStatsMap);
863 + return new DeviceEvent(PORT_STATS_UPDATED, devices.get(deviceId), null);
836 } 864 }
837 - devicePortStats.put(deviceId, statsMap); 865 +
838 - return null; // new DeviceEvent(PORT_STATS_UPDATED, devices.get(deviceId), null); 866 + /**
867 + * Calculate delta statistics by subtracting previous from new statistics.
868 + *
869 + * @param deviceId
870 + * @param prvStats
871 + * @param newStats
872 + * @return PortStatistics
873 + */
874 + public PortStatistics calcDeltaStats(DeviceId deviceId, PortStatistics prvStats, PortStatistics newStats) {
875 + // calculate time difference
876 + long deltaStatsSec, deltaStatsNano;
877 + if (newStats.durationNano() < prvStats.durationNano()) {
878 + deltaStatsNano = newStats.durationNano() - prvStats.durationNano() + TimeUnit.SECONDS.toNanos(1);
879 + deltaStatsSec = newStats.durationSec() - prvStats.durationSec() - 1L;
880 + } else {
881 + deltaStatsNano = newStats.durationNano() - prvStats.durationNano();
882 + deltaStatsSec = newStats.durationSec() - prvStats.durationSec();
883 + }
884 + DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
885 + DefaultPortStatistics deltaStats = builder.setDeviceId(deviceId)
886 + .setPort(newStats.port())
887 + .setPacketsReceived(newStats.packetsReceived() - prvStats.packetsReceived())
888 + .setPacketsSent(newStats.packetsSent() - prvStats.packetsSent())
889 + .setBytesReceived(newStats.bytesReceived() - prvStats.bytesReceived())
890 + .setBytesSent(newStats.bytesSent() - prvStats.bytesSent())
891 + .setPacketsRxDropped(newStats.packetsRxDropped() - prvStats.packetsRxDropped())
892 + .setPacketsTxDropped(newStats.packetsTxDropped() - prvStats.packetsTxDropped())
893 + .setPacketsRxErrors(newStats.packetsRxErrors() - prvStats.packetsRxErrors())
894 + .setPacketsTxErrors(newStats.packetsTxErrors() - prvStats.packetsTxErrors())
895 + .setDurationSec(deltaStatsSec)
896 + .setDurationNano(deltaStatsNano)
897 + .build();
898 + return deltaStats;
839 } 899 }
840 900
841 @Override 901 @Override
...@@ -848,6 +908,15 @@ public class GossipDeviceStore ...@@ -848,6 +908,15 @@ public class GossipDeviceStore
848 } 908 }
849 909
850 @Override 910 @Override
911 + public List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId) {
912 + Map<PortNumber, PortStatistics> portStats = devicePortDeltaStats.get(deviceId);
913 + if (portStats == null) {
914 + return Collections.emptyList();
915 + }
916 + return ImmutableList.copyOf(portStats.values());
917 + }
918 +
919 + @Override
851 public Port getPort(DeviceId deviceId, PortNumber portNumber) { 920 public Port getPort(DeviceId deviceId, PortNumber portNumber) {
852 Map<PortNumber, Port> ports = devicePorts.get(deviceId); 921 Map<PortNumber, Port> ports = devicePorts.get(deviceId);
853 return ports == null ? null : ports.get(portNumber); 922 return ports == null ? null : ports.get(portNumber);
...@@ -934,7 +1003,7 @@ public class GossipDeviceStore ...@@ -934,7 +1003,7 @@ public class GossipDeviceStore
934 markOfflineInternal(deviceId, timestamp); 1003 markOfflineInternal(deviceId, timestamp);
935 descs.clear(); 1004 descs.clear();
936 return device == null ? null : 1005 return device == null ? null :
937 - new DeviceEvent(DEVICE_REMOVED, device, null); 1006 + new DeviceEvent(DeviceEvent.Type.DEVICE_REMOVED, device, null);
938 } 1007 }
939 } 1008 }
940 1009
......