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