Gitiles
Code Review Sign In
gerrit.opencord.org / igmpproxy / 88134a45592c8c857626dd027aecb5fffc3d0c9a / . / src / main / java / org / opencord / igmpproxy / IgmpManager.java
blob: 967e46729ed7fbd1ff8ad0b9a96efec4f5fd1015 [file] [log] [blame]
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]1package org.opencord.igmpproxy;
2
3import com.google.common.collect.Maps;
4import org.apache.felix.scr.annotations.Activate;
5import org.apache.felix.scr.annotations.Component;
6import org.apache.felix.scr.annotations.Deactivate;
7import org.apache.felix.scr.annotations.Reference;
8import org.apache.felix.scr.annotations.ReferenceCardinality;
9import org.onlab.packet.EthType;
10import org.onlab.packet.Ethernet;
11import org.onlab.packet.IGMP;
12import org.onlab.packet.IGMPGroup;
13import org.onlab.packet.IGMPMembership;
14import org.onlab.packet.IGMPQuery;
15import org.onlab.packet.IPv4;
16import org.onlab.packet.Ip4Address;
17import org.onlab.packet.IpAddress;
18import org.onlab.packet.VlanId;
19import org.onosproject.core.ApplicationId;
20import org.onosproject.core.CoreService;
21import org.onosproject.incubator.net.config.basics.McastConfig;
22import org.onosproject.mastership.MastershipService;
23import org.onosproject.net.AnnotationKeys;
24import org.onosproject.net.ConnectPoint;
25import org.onosproject.net.DeviceId;
26import org.onosproject.net.Port;
27import org.onosproject.net.PortNumber;
28import org.onosproject.net.config.ConfigFactory;
29import org.onosproject.net.config.NetworkConfigEvent;
30import org.onosproject.net.config.NetworkConfigListener;
31import org.onosproject.net.config.NetworkConfigRegistry;
32import org.onosproject.net.config.basics.SubjectFactories;
33import org.onosproject.net.device.DeviceEvent;
34import org.onosproject.net.device.DeviceListener;
35import org.onosproject.net.device.DeviceService;
36import org.onosproject.net.flow.DefaultTrafficTreatment;
37import org.onosproject.net.flow.FlowRuleService;
38import org.onosproject.net.flow.criteria.Criteria;
39import org.onosproject.net.flowobjective.DefaultFilteringObjective;
40import org.onosproject.net.flowobjective.FilteringObjective;
41import org.onosproject.net.flowobjective.FlowObjectiveService;
42import org.onosproject.net.flowobjective.Objective;
43import org.onosproject.net.flowobjective.ObjectiveContext;
44import org.onosproject.net.flowobjective.ObjectiveError;
45import org.onosproject.net.mcast.McastRoute;
46import org.onosproject.net.mcast.MulticastRouteService;
47import org.onosproject.net.packet.InboundPacket;
48import org.onosproject.net.packet.PacketContext;
49import org.onosproject.net.packet.PacketProcessor;
50import org.onosproject.net.packet.PacketService;
51import org.opencord.cordconfig.access.AccessDeviceConfig;
52import org.opencord.cordconfig.access.AccessDeviceData;
53import org.slf4j.Logger;
54import org.slf4j.LoggerFactory;
55
56import java.util.ArrayList;
57import java.util.Collection;
58import java.util.Iterator;
59import java.util.List;
60import java.util.Map;
61import java.util.Set;
62import java.util.TimerTask;
63import java.util.concurrent.ConcurrentHashMap;
64import java.util.concurrent.Executors;
65import java.util.concurrent.ScheduledExecutorService;
66import java.util.concurrent.TimeUnit;
67
68/**
69 * Igmp process application, use proxy mode, support first join/ last leave , fast leave
70 * period query and keep alive, packet out igmp message to uplink port features.
71 */
72@Component(immediate = true)
73public class IgmpManager {
74
75 private static final Class<AccessDeviceConfig> CONFIG_CLASS =
76 AccessDeviceConfig.class;
77 private static final Class<IgmpproxyConfig> IGMPPROXY_CONFIG_CLASS =
78 IgmpproxyConfig.class;
79 private static final Class<IgmpproxySsmTranslateConfig> IGMPPROXY_SSM_CONFIG_CLASS =
80 IgmpproxySsmTranslateConfig.class;
81 private static final Class<McastConfig> MCAST_CONFIG_CLASS =
82 McastConfig.class;
83 public static Map<String, GroupMember> groupMemberMap = Maps.newConcurrentMap();
84 private static ApplicationId appId;
85 private static Map<DeviceId, AccessDeviceData> oltData = new ConcurrentHashMap<>();
86 private static int unSolicitedTimeout = 3; // unit is 1 sec
87 private static int keepAliveCount = 3;
88 private static int lastQueryInterval = 2; //unit is 1 sec
89 private static int lastQueryCount = 2;
90 private static boolean fastLeave = true;
91 private static boolean withRAUplink = true;
92 private static boolean withRADownlink = false;
93 private static boolean periodicQuery = true;
94 private static short mvlan = 4000;
95 private static byte igmpCos = 7;
96 public static boolean connectPointMode = true;
97 public static ConnectPoint connectPoint = null;
98
99 private final ScheduledExecutorService scheduledExecutorService =
100 Executors.newScheduledThreadPool(1);
101 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
102 protected CoreService coreService;
103 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
104 protected PacketService packetService;
105 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
106 protected MastershipService mastershipService;
107 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
108 protected FlowRuleService flowRuleService;
109 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
110 protected DeviceService deviceService;
111 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
112 protected FlowObjectiveService flowObjectiveService;
113 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
114 protected NetworkConfigRegistry networkConfig;
115 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
116 protected MulticastRouteService multicastService;
117 private IgmpPacketProcessor processor = new IgmpPacketProcessor();
118 private Logger log = LoggerFactory.getLogger(getClass());
119 private ApplicationId coreAppId;
120 private Map<Ip4Address, Ip4Address> ssmTranslateTable = new ConcurrentHashMap<>();
121 private InternalNetworkConfigListener configListener =
122 new InternalNetworkConfigListener();
123 private DeviceListener deviceListener = new InternalDeviceListener();
124 private ConfigFactory<DeviceId, AccessDeviceConfig> configFactory = null;
125 private ConfigFactory<ApplicationId, IgmpproxyConfig> igmpproxyConfigFactory =
126 new ConfigFactory<ApplicationId, IgmpproxyConfig>(
127 SubjectFactories.APP_SUBJECT_FACTORY, IGMPPROXY_CONFIG_CLASS, "igmpproxy") {
128 @Override
129 public IgmpproxyConfig createConfig() {
130 return new IgmpproxyConfig();
131 }
132 };
133 private ConfigFactory<ApplicationId, IgmpproxySsmTranslateConfig> igmpproxySsmConfigFactory =
134 new ConfigFactory<ApplicationId, IgmpproxySsmTranslateConfig>(
135 SubjectFactories.APP_SUBJECT_FACTORY, IGMPPROXY_SSM_CONFIG_CLASS, "ssmTranslate", true) {
136 @Override
137 public IgmpproxySsmTranslateConfig createConfig() {
138 return new IgmpproxySsmTranslateConfig();
139 }
140 };
141 private int maxResp = 10; //unit is 1 sec
142 private int keepAliveInterval = 120; //unit is 1 sec
143
144 public static int getUnsolicitedTimeout() {
145 return unSolicitedTimeout;
146 }
147
148 @Activate
149 protected void activate() {
150 appId = coreService.registerApplication("org.opencord.igmpproxy");
151 coreAppId = coreService.registerApplication(CoreService.CORE_APP_NAME);
152 packetService.addProcessor(processor, PacketProcessor.director(4));
153 IgmpSender.init(packetService, mastershipService);
154
155 if (networkConfig.getConfigFactory(CONFIG_CLASS) == null) {
156 configFactory =
157 new ConfigFactory<DeviceId, AccessDeviceConfig>(
158 SubjectFactories.DEVICE_SUBJECT_FACTORY, CONFIG_CLASS, "accessDevice") {
159 @Override
160 public AccessDeviceConfig createConfig() {
161 return new AccessDeviceConfig();
162 }
163 };
164 networkConfig.registerConfigFactory(configFactory);
165 }
166 networkConfig.registerConfigFactory(igmpproxySsmConfigFactory);
167 networkConfig.registerConfigFactory(igmpproxyConfigFactory);
168 networkConfig.addListener(configListener);
169
170 configListener.reconfigureNetwork(networkConfig.getConfig(appId, IGMPPROXY_CONFIG_CLASS));
171 configListener.reconfigureSsmTable(networkConfig.getConfig(appId, IGMPPROXY_SSM_CONFIG_CLASS));
172
173 networkConfig.getSubjects(DeviceId.class, AccessDeviceConfig.class).forEach(
174 subject -> {
175 AccessDeviceConfig config = networkConfig.getConfig(subject,
176 AccessDeviceConfig.class);
177 if (config != null) {
178 AccessDeviceData data = config.getAccessDevice();
179 oltData.put(data.deviceId(), data);
180 }
181 }
182 );
183
184 oltData.keySet().forEach(d->provisionDefaultFlows(d));
185 if (connectPointMode) {
186 provisionConnectPointFlows();
187 } else {
188 provisionUplinkFlows();
189 }
190
191 McastConfig config = networkConfig.getConfig(coreAppId, MCAST_CONFIG_CLASS);
192 if (config != null) {
193 mvlan = config.egressVlan().toShort();
Deepa Vaddireddy88134a42017-07-05 12:16:03 +0530[diff] [blame^]194 IgmpSender.getInstance().setMvlan(mvlan);
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]195 }
196 deviceService.addListener(deviceListener);
197 scheduledExecutorService.scheduleAtFixedRate(new IgmpProxyTimerTask(), 0, 1000, TimeUnit.MILLISECONDS);
198
199 log.info("Started");
200 }
201
202 @Deactivate
203 protected void deactivate() {
204 scheduledExecutorService.shutdown();
205
206 // de-register and null our handler
207 networkConfig.removeListener(configListener);
208 if (configFactory != null) {
209 networkConfig.unregisterConfigFactory(configFactory);
210 }
211 networkConfig.unregisterConfigFactory(igmpproxyConfigFactory);
212 networkConfig.unregisterConfigFactory(igmpproxySsmConfigFactory);
213 deviceService.removeListener(deviceListener);
214 packetService.removeProcessor(processor);
215 flowRuleService.removeFlowRulesById(appId);
216
217 log.info("Stopped");
218 }
219
220 protected Ip4Address getDeviceIp(DeviceId ofDeviceId) {
221 try {
222 String[] mgmtAddress = deviceService.getDevice(ofDeviceId)
223 .annotations().value(AnnotationKeys.MANAGEMENT_ADDRESS).split(":");
224 return Ip4Address.valueOf(mgmtAddress[0]);
225 } catch (Exception ex) {
226 log.info("No valid Ipaddress for " + ofDeviceId.toString());
227 return null;
228 }
229 }
230
231 private void processIgmpQuery(IGMPQuery igmpQuery, ConnectPoint cp, int maxResp) {
232
233 DeviceId deviceId = cp.deviceId();
234 Ip4Address gAddr = igmpQuery.getGaddr().getIp4Address();
235
236 if (maxResp >= 128) {
237 int mant = maxResp & 0xf;
238 int exp = (maxResp >> 4) & 0x7;
239 maxResp = (mant | 0x10) << (exp + 3);
240 }
241
242 maxResp = (maxResp + 5) / 10;
243
244 if (gAddr != null && !gAddr.isZero()) {
245 StateMachine.specialQuery(deviceId, gAddr, maxResp);
246 } else {
247 StateMachine.generalQuery(deviceId, maxResp);
248 }
249
250 }
251
252 private Ip4Address ssmTranslateRoute(IpAddress group) {
253 return ssmTranslateTable.get(group);
254 }
255
256 private void processIgmpReport(IGMPMembership igmpGroup, VlanId vlan, ConnectPoint cp, byte igmpType) {
257 DeviceId deviceId = cp.deviceId();
258 PortNumber portNumber = cp.port();
259
260 Ip4Address groupIp = igmpGroup.getGaddr().getIp4Address();
261 if (!groupIp.isMulticast()) {
262 log.info(groupIp.toString() + " is not a valid group address");
263 return;
264 }
265 Ip4Address srcIp = getDeviceIp(deviceId);
266
267 byte recordType = igmpGroup.getRecordType();
268 boolean join = false;
269
270 ArrayList<Ip4Address> sourceList = new ArrayList<>();
271
272 if (igmpGroup.getSources().size() > 0) {
273 igmpGroup.getSources().forEach(source -> sourceList.add(source.getIp4Address()));
274 if (recordType == IGMPMembership.CHANGE_TO_EXCLUDE_MODE ||
275 recordType == IGMPMembership.MODE_IS_EXCLUDE ||
276 recordType == IGMPMembership.BLOCK_OLD_SOURCES) {
277 join = false;
278 } else if (recordType == IGMPMembership.CHANGE_TO_INCLUDE_MODE ||
279 recordType == IGMPMembership.MODE_IS_INCLUDE ||
280 recordType == IGMPMembership.ALLOW_NEW_SOURCES) {
281 join = true;
282 }
283 } else {
284 IpAddress src = ssmTranslateRoute(groupIp);
285 if (src == null) {
286 log.info("no ssm translate for group " + groupIp.toString());
287 return;
288 }
289 sourceList.add(src.getIp4Address());
290 if (recordType == IGMPMembership.CHANGE_TO_EXCLUDE_MODE ||
291 recordType == IGMPMembership.MODE_IS_EXCLUDE ||
292 recordType == IGMPMembership.BLOCK_OLD_SOURCES) {
293 join = true;
294 } else if (recordType == IGMPMembership.CHANGE_TO_INCLUDE_MODE ||
295 recordType == IGMPMembership.MODE_IS_INCLUDE ||
296 recordType == IGMPMembership.ALLOW_NEW_SOURCES) {
297 join = false;
298 }
299 }
300 String groupMemberKey = GroupMember.getkey(groupIp, deviceId, portNumber);
301 GroupMember groupMember = groupMemberMap.get(groupMemberKey);
302
303 if (join) {
304 if (groupMember == null) {
305 if (igmpType == IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT) {
306 groupMember = new GroupMember(groupIp, vlan, deviceId, portNumber, true);
307 } else {
308 groupMember = new GroupMember(groupIp, vlan, deviceId, portNumber, false);
309 }
310 StateMachine.join(deviceId, groupIp, srcIp);
311 groupMemberMap.put(groupMemberKey, groupMember);
312 groupMember.updateList(recordType, sourceList);
313 groupMember.getSourceList().forEach(source -> multicastService.addSink(
314 new McastRoute(source, groupIp, McastRoute.Type.IGMP), cp));
315 }
316 groupMember.resetAllTimers();
317 groupMember.updateList(recordType, sourceList);
318 groupMember.setLeave(false);
319 } else {
320 if (groupMember == null) {
321 log.info("receive leave but no instance, group " + groupIp.toString() +
322 " device:" + deviceId.toString() + " port:" + portNumber.toString());
323 return;
324 } else {
325 groupMember.setLeave(true);
326 if (fastLeave) {
327 leaveAction(groupMember);
328 } else {
329 sendQuery(groupMember);
330 }
331 }
332 }
333 }
334
335 private void leaveAction(GroupMember groupMember) {
336 ConnectPoint cp = new ConnectPoint(groupMember.getDeviceId(), groupMember.getPortNumber());
337 StateMachine.leave(groupMember.getDeviceId(), groupMember.getGroupIp());
338 groupMember.getSourceList().forEach(source -> multicastService.removeSink(
339 new McastRoute(source, groupMember.getGroupIp(),
340 McastRoute.Type.IGMP), cp));
341 groupMemberMap.remove(groupMember.getId());
342 }
343
344 private void sendQuery(GroupMember groupMember) {
345 Ethernet ethpkt;
346 Ip4Address srcIp = getDeviceIp(groupMember.getDeviceId());
347 if (groupMember.getv2()) {
348 ethpkt = IgmpSender.getInstance().buildIgmpV2Query(groupMember.getGroupIp(), srcIp);
349 } else {
350 ethpkt = IgmpSender.getInstance().buildIgmpV3Query(groupMember.getGroupIp(), srcIp);
351 }
352 IgmpSender.getInstance().sendIgmpPacket(ethpkt, groupMember.getDeviceId(), groupMember.getPortNumber());
353 }
354
355 private void processFilterObjective(DeviceId devId, Port port, boolean remove) {
356
357 //TODO migrate to packet requests when packet service uses filtering objectives
358 DefaultFilteringObjective.Builder builder = DefaultFilteringObjective.builder();
359
360 builder = remove ? builder.deny() : builder.permit();
361
362 FilteringObjective igmp = builder
363 .withKey(Criteria.matchInPort(port.number()))
364 .addCondition(Criteria.matchEthType(EthType.EtherType.IPV4.ethType()))
365 .addCondition(Criteria.matchIPProtocol(IPv4.PROTOCOL_IGMP))
366 .withMeta(DefaultTrafficTreatment.builder().setOutput(PortNumber.CONTROLLER).build())
367 .fromApp(appId)
368 .withPriority(10000)
369 .add(new ObjectiveContext() {
370 @Override
371 public void onSuccess(Objective objective) {
372 log.info("IgmpProxy filter for {} on {} installed.",
373 devId, port);
374 }
375
376 @Override
377 public void onError(Objective objective, ObjectiveError error) {
378 log.info("IgmpProxy filter for {} on {} failed because {}.",
379 devId, port, error);
380 }
381 });
382
383 flowObjectiveService.filter(devId, igmp);
384 }
385
386 /**
387 * Packet processor responsible for forwarding packets along their paths.
388 */
389 private class IgmpPacketProcessor implements PacketProcessor {
390 @Override
391 public void process(PacketContext context) {
392
393 try {
394 InboundPacket pkt = context.inPacket();
395 Ethernet ethPkt = pkt.parsed();
396 if (ethPkt == null) {
397 return;
398 }
399
400 if (ethPkt.getEtherType() != Ethernet.TYPE_IPV4) {
401 return;
402 }
403
404 IPv4 ipv4Pkt = (IPv4) ethPkt.getPayload();
405
406 if (ipv4Pkt.getProtocol() != IPv4.PROTOCOL_IGMP) {
407 return;
408 }
409
410 short vlan = ethPkt.getVlanID();
411 DeviceId deviceId = pkt.receivedFrom().deviceId();
412
413 if (oltData.get(deviceId) == null) {
414 log.error("Device not registered in netcfg :" + deviceId.toString());
415 return;
416 }
417
418 IGMP igmp = (IGMP) ipv4Pkt.getPayload();
419 switch (igmp.getIgmpType()) {
420 case IGMP.TYPE_IGMPV3_MEMBERSHIP_QUERY:
421 //Discard Query from OLT’s non-uplink port’s
422 if (!pkt.receivedFrom().port().equals(getDeviceUplink(deviceId))) {
423 log.info("IGMP Picked up query from non-uplink port");
424 return;
425 }
426
427 processIgmpQuery((IGMPQuery) igmp.getGroups().get(0), pkt.receivedFrom(),
428 0xff & igmp.getMaxRespField());
429 break;
430 case IGMP.TYPE_IGMPV1_MEMBERSHIP_REPORT:
431 log.debug("IGMP version 1 message types are not currently supported.");
432 break;
433 case IGMP.TYPE_IGMPV3_MEMBERSHIP_REPORT:
434 case IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT:
435 case IGMP.TYPE_IGMPV2_LEAVE_GROUP:
436 //Discard join/leave from OLT’s uplink port’s
437 if (pkt.receivedFrom().port().equals(getDeviceUplink(deviceId))) {
438 log.info("IGMP Picked up join/leave from the olt uplink port");
439 return;
440 }
441
442 Iterator<IGMPGroup> itr = igmp.getGroups().iterator();
443 while (itr.hasNext()) {
444 IGMPGroup group = itr.next();
445 if (group instanceof IGMPMembership) {
446 processIgmpReport((IGMPMembership) group, VlanId.vlanId(vlan),
447 pkt.receivedFrom(), igmp.getIgmpType());
448 } else if (group instanceof IGMPQuery) {
449 IGMPMembership mgroup;
450 mgroup = new IGMPMembership(group.getGaddr().getIp4Address());
451 mgroup.setRecordType(igmp.getIgmpType() == IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT ?
452 IGMPMembership.MODE_IS_EXCLUDE : IGMPMembership.MODE_IS_INCLUDE);
453 processIgmpReport(mgroup, VlanId.vlanId(vlan),
454 pkt.receivedFrom(), igmp.getIgmpType());
455 }
456 }
457 break;
458
459 default:
460 log.info("wrong IGMP v3 type:" + igmp.getIgmpType());
461 break;
462 }
463
464 } catch (Exception ex) {
465 log.error("igmp process error : " + ex.toString());
466 ex.printStackTrace();
467 }
468 }
469 }
470
471 private class IgmpProxyTimerTask extends TimerTask {
472 public void run() {
473 try {
474 IgmpTimer.timeOut1s();
475 queryMembers();
476 } catch (Exception ex) {
477 log.warn("Igmp timer task error : {}", ex.getMessage());
478 }
479 }
480
481 private void queryMembers() {
482 GroupMember groupMember;
483 Set groupMemberSet = groupMemberMap.entrySet();
484 Iterator itr = groupMemberSet.iterator();
485 while (itr.hasNext()) {
486 Map.Entry entry = (Map.Entry) itr.next();
487 groupMember = (GroupMember) entry.getValue();
488 DeviceId did = groupMember.getDeviceId();
489 if (mastershipService.isLocalMaster(did)) {
490 if (groupMember.isLeave()) {
491 lastQuery(groupMember);
492 } else if (periodicQuery) {
493 periodicQuery(groupMember);
494 }
495 }
496 }
497 }
498
499 private void lastQuery(GroupMember groupMember) {
500 if (groupMember.getLastQueryInterval() < lastQueryInterval) {
501 groupMember.lastQueryInterval(true); // count times
502 } else if (groupMember.getLastQueryCount() < lastQueryCount - 1) {
503 sendQuery(groupMember);
504 groupMember.lastQueryInterval(false); // reset count number
505 groupMember.lastQueryCount(true); //count times
506 } else if (groupMember.getLastQueryCount() == lastQueryCount - 1) {
507 leaveAction(groupMember);
508 }
509 }
510
511 private void periodicQuery(GroupMember groupMember) {
512 if (groupMember.getKeepAliveQueryInterval() < keepAliveInterval) {
513 groupMember.keepAliveInterval(true);
514 } else if (groupMember.getKeepAliveQueryCount() < keepAliveCount) {
515 sendQuery(groupMember);
516 groupMember.keepAliveInterval(false);
517 groupMember.keepAliveQueryCount(true);
518 } else if (groupMember.getKeepAliveQueryCount() == keepAliveCount) {
519 leaveAction(groupMember);
520 }
521 }
522
523 }
524
525 public static PortNumber getDeviceUplink(DeviceId devId) {
526 return oltData.get(devId).uplink();
527 }
528
529 private void processFilterObjective(DeviceId devId, PortNumber port, boolean remove) {
530 //TODO migrate to packet requests when packet service uses filtering objectives
531 DefaultFilteringObjective.Builder builder = DefaultFilteringObjective.builder();
532
533 builder = remove ? builder.deny() : builder.permit();
534
535 FilteringObjective igmp = builder
536 .withKey(Criteria.matchInPort(port))
537 .addCondition(Criteria.matchEthType(EthType.EtherType.IPV4.ethType()))
538 .addCondition(Criteria.matchIPProtocol(IPv4.PROTOCOL_IGMP))
539 .withMeta(DefaultTrafficTreatment.builder().setOutput(PortNumber.CONTROLLER).build())
540 .fromApp(appId)
541 .withPriority(10000)
542 .add(new ObjectiveContext() {
543 @Override
544 public void onSuccess(Objective objective) {
545 log.info("IgmpProxy filter for {} on {} installed.",
546 devId, port);
547 }
548
549 @Override
550 public void onError(Objective objective, ObjectiveError error) {
551 log.info("IgmpProxy filter for {} on {} failed because {}.",
552 devId, port, error);
553 }
554 });
555
556 flowObjectiveService.filter(devId, igmp);
557 }
558
559 private boolean isConnectPoint(DeviceId device, PortNumber port) {
560 return (connectPointMode && connectPoint.deviceId().equals(device)
561 && connectPoint.port().equals(port));
562 }
563 private boolean isUplink(DeviceId device, PortNumber port) {
564 return ((!connectPointMode) && oltData.containsKey(device)
565 && oltData.get(device).uplink().equals(port));
566 }
567
568 private class InternalDeviceListener implements DeviceListener {
569 @Override
570 public void event(DeviceEvent event) {
571 DeviceId devId = event.subject().id();
572 PortNumber port;
573 if (oltData.get(devId) == null) {
574 return;
575 }
576 switch (event.type()) {
577
578 case DEVICE_ADDED:
579 case DEVICE_UPDATED:
580 case DEVICE_REMOVED:
581 case DEVICE_SUSPENDED:
582 case DEVICE_AVAILABILITY_CHANGED:
583 case PORT_STATS_UPDATED:
584 break;
585 case PORT_ADDED:
586 port = event.port().number();
587 if (oltData.containsKey(devId) && !isUplink(devId, port) && !isConnectPoint(devId, port)) {
588 processFilterObjective(devId, port, false);
589 } else if (isUplink(devId, port)) {
590 provisionUplinkFlows();
591 } else if (isConnectPoint(devId, port)) {
592 provisionConnectPointFlows();
593 }
594 break;
595 case PORT_UPDATED:
596 port = event.port().number();
597 if (oltData.containsKey(devId) && !isUplink(devId, port) && !isConnectPoint(devId, port)) {
598 if (event.port().isEnabled()) {
599 processFilterObjective(devId, port, false);
600 } else {
601 processFilterObjective(devId, port, true);
602 }
603 } else if (isUplink(devId, port)) {
604 if (event.port().isEnabled()) {
605 provisionUplinkFlows(devId);
606 } else {
607 processFilterObjective(devId, port, true);
608 }
609 } else if (isConnectPoint(devId, port)) {
610 if (event.port().isEnabled()) {
611 provisionConnectPointFlows();
612 } else {
613 unprovisionConnectPointFlows();
614 }
615 }
616 break;
617 case PORT_REMOVED:
618 port = event.port().number();
619 processFilterObjective(devId, port, true);
620 break;
621 default:
622 log.info("Unknown device event {}", event.type());
623 break;
624 }
625 }
626
627 @Override
628 public boolean isRelevant(DeviceEvent event) {
629 return true;
630 }
631 }
632
633 private class InternalNetworkConfigListener implements NetworkConfigListener {
634
635 private void reconfigureNetwork(IgmpproxyConfig cfg) {
636 IgmpproxyConfig newCfg;
637 if (cfg == null) {
638 newCfg = new IgmpproxyConfig();
639 } else {
640 newCfg = cfg;
641 }
642
643 unSolicitedTimeout = newCfg.unsolicitedTimeOut();
644 maxResp = newCfg.maxResp();
645 keepAliveInterval = newCfg.keepAliveInterval();
646 keepAliveCount = newCfg.keepAliveCount();
647 lastQueryInterval = newCfg.lastQueryInterval();
648 lastQueryCount = newCfg.lastQueryCount();
649 withRAUplink = newCfg.withRAUplink();
650 withRADownlink = newCfg.withRADownlink();
651 igmpCos = newCfg.igmpCos();
652 periodicQuery = newCfg.periodicQuery();
653 fastLeave = newCfg.fastLeave();
654
655 connectPoint = newCfg.connectPoint();
656 if (connectPointMode != newCfg.connectPointMode()) {
657 connectPointMode = newCfg.connectPointMode();
658 if (connectPointMode) {
659 unprovisionUplinkFlows();
660 provisionConnectPointFlows();
661 } else {
662 unprovisionConnectPointFlows();
663 provisionUplinkFlows();
664 }
665 }
666 if (connectPoint != null) {
667 log.info("connect point :" + connectPoint.toString());
668 }
669 log.info(" mode: " + connectPointMode);
670
671 IgmpSender.getInstance().setIgmpCos(igmpCos);
672 IgmpSender.getInstance().setMaxResp(maxResp);
673 IgmpSender.getInstance().setMvlan(mvlan);
674 IgmpSender.getInstance().setWithRADownlink(withRADownlink);
675 IgmpSender.getInstance().setWithRAUplink(withRAUplink);
676
677 }
678
679 public void reconfigureSsmTable(IgmpproxySsmTranslateConfig cfg) {
680 if (cfg == null) {
681 return;
682 }
683 Collection<McastRoute> translations = cfg.getSsmTranslations();
684 for (McastRoute route : translations) {
685 ssmTranslateTable.put(route.group().getIp4Address(), route.source().getIp4Address());
686 }
687 }
688
689 @Override
690 public void event(NetworkConfigEvent event) {
691 switch (event.type()) {
692 case CONFIG_ADDED:
693 case CONFIG_UPDATED:
694 if (event.configClass().equals(CONFIG_CLASS)) {
695 AccessDeviceConfig config =
696 networkConfig.getConfig((DeviceId) event.subject(), CONFIG_CLASS);
697 if (config != null) {
698 oltData.put(config.getAccessDevice().deviceId(), config.getAccessDevice());
699 provisionDefaultFlows((DeviceId) event.subject());
700 provisionUplinkFlows((DeviceId) event.subject());
701 }
702 }
703
704 if (event.configClass().equals(IGMPPROXY_CONFIG_CLASS)) {
705 IgmpproxyConfig config = networkConfig.getConfig(appId, IGMPPROXY_CONFIG_CLASS);
706 if (config != null) {
707 reconfigureNetwork(config);
708 }
709 }
710
711 if (event.configClass().equals(IGMPPROXY_SSM_CONFIG_CLASS)) {
712 IgmpproxySsmTranslateConfig config = networkConfig.getConfig(appId, IGMPPROXY_SSM_CONFIG_CLASS);
713 if (config != null) {
714 reconfigureSsmTable(config);
715 }
716 }
717
718 if (event.configClass().equals(MCAST_CONFIG_CLASS)) {
719 McastConfig config = networkConfig.getConfig(coreAppId, MCAST_CONFIG_CLASS);
720 if (config != null && mvlan != config.egressVlan().toShort()) {
721 mvlan = config.egressVlan().toShort();
Deepa Vaddireddy88134a42017-07-05 12:16:03 +0530[diff] [blame^]722 IgmpSender.getInstance().setMvlan(mvlan);
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]723 groupMemberMap.values().forEach(m -> leaveAction(m));
724 }
725 }
726
727 log.info("Reconfigured");
728 break;
729 case CONFIG_REGISTERED:
730 case CONFIG_UNREGISTERED:
731 break;
732 case CONFIG_REMOVED:
733 if (event.configClass().equals(CONFIG_CLASS)) {
734 oltData.remove(event.subject());
735 }
736
737 default:
738 break;
739 }
740 }
741 }
742
743 private void provisionDefaultFlows(DeviceId deviceId) {
744 List<Port> ports = deviceService.getPorts(deviceId);
745 ports.stream()
746 .filter(p -> (!oltData.get(p.element().id()).uplink().equals(p.number()) && p.isEnabled()))
747 .forEach(p -> processFilterObjective((DeviceId) p.element().id(), p.number(), false));
748 }
749
750 private void provisionUplinkFlows(DeviceId deviceId) {
751 if (connectPointMode) {
752 return;
753 }
754
755 processFilterObjective(deviceId, oltData.get(deviceId).uplink(), false);
756 }
757
758 private void provisionUplinkFlows() {
759 if (connectPointMode) {
760 return;
761 }
762
763 oltData.keySet().forEach(deviceId ->provisionUplinkFlows(deviceId));
764 }
765 private void unprovisionUplinkFlows() {
766 oltData.keySet().forEach(deviceId ->
767 processFilterObjective(deviceId, oltData.get(deviceId).uplink(), true));
768 }
769
770 private void provisionConnectPointFlows() {
771 if ((!connectPointMode) || connectPoint == null) {
772 return;
773 }
774
775 processFilterObjective(connectPoint.deviceId(), connectPoint.port(), false);
776 }
777 private void unprovisionConnectPointFlows() {
778 if (connectPoint == null) {
779 return;
780 }
781 processFilterObjective(connectPoint.deviceId(), connectPoint.port(), true);
782 }
783}