Thomas Vachuska
Committed by Gerrit Code Review

Finished implementation of edge port manager using topology event async notifications.

Change-Id: Ide0eb947ba6400dafe11dac73af1466aaf0ce451
...@@ -23,7 +23,9 @@ import java.nio.ByteBuffer; ...@@ -23,7 +23,9 @@ import java.nio.ByteBuffer;
23 import java.util.Optional; 23 import java.util.Optional;
24 24
25 /** 25 /**
26 - * Service for interacting with an inventory of network edge ports. 26 + * Service for interacting with an inventory of network edge ports. A port
27 + * is considered an edge port if it is an active port and does not have an
28 + * infrastructure link associated with it.
27 */ 29 */
28 public interface EdgePortService { 30 public interface EdgePortService {
29 31
......
...@@ -54,6 +54,7 @@ import java.util.Map; ...@@ -54,6 +54,7 @@ import java.util.Map;
54 import java.util.Optional; 54 import java.util.Optional;
55 import java.util.Set; 55 import java.util.Set;
56 56
57 +import static org.onosproject.net.device.DeviceEvent.Type.*;
57 import static org.onosproject.net.edge.EdgePortEvent.Type.EDGE_PORT_ADDED; 58 import static org.onosproject.net.edge.EdgePortEvent.Type.EDGE_PORT_ADDED;
58 import static org.onosproject.net.edge.EdgePortEvent.Type.EDGE_PORT_REMOVED; 59 import static org.onosproject.net.edge.EdgePortEvent.Type.EDGE_PORT_REMOVED;
59 import static org.slf4j.LoggerFactory.getLogger; 60 import static org.slf4j.LoggerFactory.getLogger;
...@@ -65,17 +66,17 @@ import static org.slf4j.LoggerFactory.getLogger; ...@@ -65,17 +66,17 @@ import static org.slf4j.LoggerFactory.getLogger;
65 @Service 66 @Service
66 public class EdgeManager implements EdgePortService { 67 public class EdgeManager implements EdgePortService {
67 68
68 - private final ListenerRegistry<EdgePortEvent, EdgePortListener>
69 - listenerRegistry = new ListenerRegistry<>();
70 -
71 private final Logger log = getLogger(getClass()); 69 private final Logger log = getLogger(getClass());
72 70
73 private Topology topology; 71 private Topology topology;
74 72
75 - private final TopologyListener topologyListener = new InnerTopologyListener();
76 -
77 private final Map<DeviceId, Set<ConnectPoint>> connectionPoints = Maps.newConcurrentMap(); 73 private final Map<DeviceId, Set<ConnectPoint>> connectionPoints = Maps.newConcurrentMap();
78 74
75 + private final ListenerRegistry<EdgePortEvent, EdgePortListener>
76 + listenerRegistry = new ListenerRegistry<>();
77 +
78 + private final TopologyListener topologyListener = new InnerTopologyListener();
79 +
79 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 80 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
80 protected EventDeliveryService eventDispatcher; 81 protected EventDeliveryService eventDispatcher;
81 82
...@@ -102,19 +103,20 @@ public class EdgeManager implements EdgePortService { ...@@ -102,19 +103,20 @@ public class EdgeManager implements EdgePortService {
102 log.info("Stopped"); 103 log.info("Stopped");
103 } 104 }
104 105
106 + @Override
105 public boolean isEdgePoint(ConnectPoint point) { 107 public boolean isEdgePoint(ConnectPoint point) {
106 return !topologyService.isInfrastructure(topologyService.currentTopology(), point); 108 return !topologyService.isInfrastructure(topologyService.currentTopology(), point);
107 } 109 }
108 110
111 + @Override
109 public Iterable<ConnectPoint> getEdgePoints() { 112 public Iterable<ConnectPoint> getEdgePoints() {
110 - //TODO if this is called before any notifications need to populate structure
111 ImmutableSet.Builder<ConnectPoint> builder = ImmutableSet.builder(); 113 ImmutableSet.Builder<ConnectPoint> builder = ImmutableSet.builder();
112 connectionPoints.forEach((k, v) -> v.forEach(builder::add)); 114 connectionPoints.forEach((k, v) -> v.forEach(builder::add));
113 return builder.build(); 115 return builder.build();
114 } 116 }
115 117
118 + @Override
116 public Iterable<ConnectPoint> getEdgePoints(DeviceId deviceId) { 119 public Iterable<ConnectPoint> getEdgePoints(DeviceId deviceId) {
117 - //TODO if this is called before any notifications need to populate structure
118 ImmutableSet.Builder<ConnectPoint> builder = ImmutableSet.builder(); 120 ImmutableSet.Builder<ConnectPoint> builder = ImmutableSet.builder();
119 Set<ConnectPoint> set = connectionPoints.get(deviceId); 121 Set<ConnectPoint> set = connectionPoints.get(deviceId);
120 if (set != null) { 122 if (set != null) {
...@@ -123,6 +125,7 @@ public class EdgeManager implements EdgePortService { ...@@ -123,6 +125,7 @@ public class EdgeManager implements EdgePortService {
123 return builder.build(); 125 return builder.build();
124 } 126 }
125 127
128 + @Override
126 public void emitPacket(ByteBuffer data, Optional<TrafficTreatment> treatment) { 129 public void emitPacket(ByteBuffer data, Optional<TrafficTreatment> treatment) {
127 TrafficTreatment.Builder builder = treatment.isPresent() ? 130 TrafficTreatment.Builder builder = treatment.isPresent() ?
128 DefaultTrafficTreatment.builder(treatment.get()) : 131 DefaultTrafficTreatment.builder(treatment.get()) :
...@@ -130,13 +133,13 @@ public class EdgeManager implements EdgePortService { ...@@ -130,13 +133,13 @@ public class EdgeManager implements EdgePortService {
130 getEdgePoints().forEach(p -> packetService.emit(packet(builder, p, data))); 133 getEdgePoints().forEach(p -> packetService.emit(packet(builder, p, data)));
131 } 134 }
132 135
136 + @Override
133 public void emitPacket(DeviceId deviceId, ByteBuffer data, 137 public void emitPacket(DeviceId deviceId, ByteBuffer data,
134 Optional<TrafficTreatment> treatment) { 138 Optional<TrafficTreatment> treatment) {
135 TrafficTreatment.Builder builder = treatment.isPresent() ? 139 TrafficTreatment.Builder builder = treatment.isPresent() ?
136 DefaultTrafficTreatment.builder(treatment.get()) : 140 DefaultTrafficTreatment.builder(treatment.get()) :
137 DefaultTrafficTreatment.builder(); 141 DefaultTrafficTreatment.builder();
138 getEdgePoints(deviceId).forEach(p -> packetService.emit(packet(builder, p, data))); 142 getEdgePoints(deviceId).forEach(p -> packetService.emit(packet(builder, p, data)));
139 -
140 } 143 }
141 144
142 private OutboundPacket packet(TrafficTreatment.Builder builder, ConnectPoint point, ByteBuffer data) { 145 private OutboundPacket packet(TrafficTreatment.Builder builder, ConnectPoint point, ByteBuffer data) {
...@@ -144,15 +147,19 @@ public class EdgeManager implements EdgePortService { ...@@ -144,15 +147,19 @@ public class EdgeManager implements EdgePortService {
144 return new DefaultOutboundPacket(point.deviceId(), builder.build(), data); 147 return new DefaultOutboundPacket(point.deviceId(), builder.build(), data);
145 } 148 }
146 149
150 + @Override
147 public void addListener(EdgePortListener listener) { 151 public void addListener(EdgePortListener listener) {
148 listenerRegistry.addListener(listener); 152 listenerRegistry.addListener(listener);
149 } 153 }
150 154
155 + @Override
151 public void removeListener(EdgePortListener listener) { 156 public void removeListener(EdgePortListener listener) {
152 listenerRegistry.removeListener(listener); 157 listenerRegistry.removeListener(listener);
153 } 158 }
154 159
155 160
161 + // Internal listener for topo events used to keep our edge-port cache
162 + // up to date.
156 private class InnerTopologyListener implements TopologyListener { 163 private class InnerTopologyListener implements TopologyListener {
157 @Override 164 @Override
158 public void event(TopologyEvent event) { 165 public void event(TopologyEvent event) {
...@@ -161,12 +168,9 @@ public class EdgeManager implements EdgePortService { ...@@ -161,12 +168,9 @@ public class EdgeManager implements EdgePortService {
161 if (triggers != null) { 168 if (triggers != null) {
162 triggers.forEach(reason -> { 169 triggers.forEach(reason -> {
163 if (reason instanceof DeviceEvent) { 170 if (reason instanceof DeviceEvent) {
164 - //TODO spuriously catches events not handled in the handler method
165 processDeviceEvent((DeviceEvent) reason); 171 processDeviceEvent((DeviceEvent) reason);
166 } else if (reason instanceof LinkEvent) { 172 } else if (reason instanceof LinkEvent) {
167 processLinkEvent((LinkEvent) reason); 173 processLinkEvent((LinkEvent) reason);
168 - } else {
169 - System.out.println(reason.toString());
170 } 174 }
171 }); 175 });
172 } else { 176 } else {
...@@ -175,12 +179,13 @@ public class EdgeManager implements EdgePortService { ...@@ -175,12 +179,13 @@ public class EdgeManager implements EdgePortService {
175 } 179 }
176 } 180 }
177 181
178 - 182 + // Initial loading of the edge port cache.
179 private void loadAllEdgePorts() { 183 private void loadAllEdgePorts() {
180 deviceService.getDevices().forEach(d -> deviceService.getPorts(d.id()) 184 deviceService.getDevices().forEach(d -> deviceService.getPorts(d.id())
181 .forEach(p -> addEdgePort(new ConnectPoint(d.id(), p.number())))); 185 .forEach(p -> addEdgePort(new ConnectPoint(d.id(), p.number()))));
182 } 186 }
183 187
188 + // Processes a link event by adding or removing its end-points in our cache.
184 private void processLinkEvent(LinkEvent event) { 189 private void processLinkEvent(LinkEvent event) {
185 if (event.type() == LinkEvent.Type.LINK_ADDED) { 190 if (event.type() == LinkEvent.Type.LINK_ADDED) {
186 removeEdgePort(event.subject().src()); 191 removeEdgePort(event.subject().src());
...@@ -189,22 +194,37 @@ public class EdgeManager implements EdgePortService { ...@@ -189,22 +194,37 @@ public class EdgeManager implements EdgePortService {
189 addEdgePort(event.subject().src()); 194 addEdgePort(event.subject().src());
190 addEdgePort(event.subject().dst()); 195 addEdgePort(event.subject().dst());
191 } 196 }
192 -
193 } 197 }
194 198
199 + // Processes a device event by adding or removing its end-points in our cache.
195 private void processDeviceEvent(DeviceEvent event) { 200 private void processDeviceEvent(DeviceEvent event) {
196 - 201 + DeviceEvent.Type type = event.type();
197 - if (event.type() == DeviceEvent.Type.PORT_ADDED) { 202 + DeviceId id = event.subject().id();
198 - addEdgePort(new ConnectPoint(event.subject().id(), event.port().number())); 203 +
199 - } else if (event.type() == DeviceEvent.Type.PORT_REMOVED) { 204 + if (type == DEVICE_ADDED ||
200 - removeEdgePort(new ConnectPoint(event.subject().id(), event.port().number())); 205 + type == DEVICE_AVAILABILITY_CHANGED && deviceService.isAvailable(id)) {
201 - } 206 + // When device is added or becomes available, add all its ports
202 - } 207 + deviceService.getPorts(event.subject().id())
203 - 208 + .forEach(p -> addEdgePort(new ConnectPoint(id, p.number())));
209 + } else if (type == DEVICE_REMOVED ||
210 + type == DEVICE_AVAILABILITY_CHANGED && !deviceService.isAvailable(id)) {
211 + // When device is removed or becomes unavailable, remove all its ports
212 + deviceService.getPorts(event.subject().id())
213 + .forEach(p -> removeEdgePort(new ConnectPoint(id, p.number())));
214 + connectionPoints.remove(id);
215 +
216 + } else if (type == DeviceEvent.Type.PORT_ADDED ||
217 + type == PORT_UPDATED && event.port().isEnabled()) {
218 + addEdgePort(new ConnectPoint(id, event.port().number()));
219 + } else if (type == DeviceEvent.Type.PORT_REMOVED ||
220 + type == PORT_UPDATED && !event.port().isEnabled()) {
221 + removeEdgePort(new ConnectPoint(id, event.port().number()));
222 + }
223 + }
224 +
225 + // Adds the specified connection point to the edge points if needed.
204 private void addEdgePort(ConnectPoint point) { 226 private void addEdgePort(ConnectPoint point) {
205 - //TODO case of link removed and one of the end ports removed in same topo cycle 227 + if (!topologyService.isInfrastructure(topology, point) && !point.port().isLogical()) {
206 - //TODO pt2. resulting behavior will be adding a non-existent edge to the set
207 - if (!topologyService.isInfrastructure(topology, point)) {
208 Set<ConnectPoint> set = connectionPoints.get(point.deviceId()); 228 Set<ConnectPoint> set = connectionPoints.get(point.deviceId());
209 if (set == null) { 229 if (set == null) {
210 set = Sets.newConcurrentHashSet(); 230 set = Sets.newConcurrentHashSet();
...@@ -214,13 +234,11 @@ public class EdgeManager implements EdgePortService { ...@@ -214,13 +234,11 @@ public class EdgeManager implements EdgePortService {
214 eventDispatcher.post(new EdgePortEvent(EDGE_PORT_ADDED, point)); 234 eventDispatcher.post(new EdgePortEvent(EDGE_PORT_ADDED, point));
215 } 235 }
216 } 236 }
217 -
218 } 237 }
219 238
239 + // Removes the specified connection point from the edge points.
220 private void removeEdgePort(ConnectPoint point) { 240 private void removeEdgePort(ConnectPoint point) {
221 - //TODO need to check that points still exist IE when a link and port are removed 241 + if (!point.port().isLogical()) {
222 - //TODO pt2 and both events are captures in the same topo update
223 - if (!topologyService.isInfrastructure(topology, point)) {
224 Set<ConnectPoint> set = connectionPoints.get(point.deviceId()); 242 Set<ConnectPoint> set = connectionPoints.get(point.deviceId());
225 if (set == null) { 243 if (set == null) {
226 return; 244 return;
...@@ -232,6 +250,5 @@ public class EdgeManager implements EdgePortService { ...@@ -232,6 +250,5 @@ public class EdgeManager implements EdgePortService {
232 connectionPoints.remove(point.deviceId()); 250 connectionPoints.remove(point.deviceId());
233 } 251 }
234 } 252 }
235 -
236 } 253 }
237 } 254 }
......